Skip to content
Sarj Sarj Sarj Standards

store-get-delegates-to-bulk-read

A store singleton operation should reuse its equivalent bulk implementation.

SARJ421 Rule source Tests Catalog JSON

Why

Independent singleton and bulk queries can drift in filtering, mutation semantics, row conversion, authorization, timestamps, and missing-row behavior while maintaining two database access paths.

Fix

Delegate only when tenant, authorization, consistency, cache, lock, transaction, conversion, and missing-row semantics match. Otherwise add an exact SARJ421 suppression naming the concrete semantic difference.

Examples

Before — flagged Do not maintain a second singleton query path
app/user_store.py
class UserStore:
    async def get(self, user_id: UserId) -> User | None:
        return await self.fetchrow("SELECT * FROM users WHERE id = %s", user_id)


    async def get_many(self, user_ids: list[UserId]) -> list[User]:
        return await self.fetch("SELECT * FROM users WHERE id = ANY(%s)", user_ids)
After — preferred Delegate the singleton read to the bulk implementation
app/user_store.py
class UserStore:
    async def get(self, user_id: UserId) -> User | None:
        rows = await self.get_by_ids([user_id])
        return rows.get(user_id)


    async def get_by_ids(self, user_ids: list[UserId]) -> dict[UserId, User]:
        return await self.query_many(user_ids)
Before — flagged Do not maintain a second singleton mutation path
app/task_store.py
class TaskStore:
    async def set_to_failed(self, task_id: str) -> Task:
        return await self._update_status(task_id, TaskStatus.FAILED)


    async def set_many_to_status(
        self, task_ids: Collection[str], status: TaskStatus
    ) -> list[Task]:
        return await self._update_many(task_ids, status)
After — preferred Route singleton mutations through the bulk primitive
app/task_store.py
class TaskStore:
    async def set_to_failed(self, task_id: str) -> Task:
        rows = await self.set_many_to_status([task_id], TaskStatus.FAILED)
        return rows[0]


    async def set_many_to_status(
        self, task_ids: Collection[str], status: TaskStatus
    ) -> list[Task]:
        return await self._update_many(task_ids, status)

Formerly: get-delegates-to-get-many