Business Logic in Repositories

Why business rules accumulate in the repository layer — and why this is harder to unwind than it first appears.

The Repository's Job

A repository has one responsibility: abstract the persistence mechanism. It translates between the domain model and the storage layer. It does not make business decisions.

Business logic in a repository is logic that cannot be tested without a database, cannot be changed without modifying data access code, and is invisible to anyone reading the application or domain layers.

How It Gets There

The Access Control Pattern

The most common form. A service needs to check whether a caller has permission to perform an operation on a resource. Rather than handling this in the application layer using domain rules, the check is delegated to the repository:

# In a repository — but this is a business decision
async def check_user_access(
    self,
    session: AsyncSession,
    resource_id: UUID,
    user_id: UUID,
    operation: AccessOperation,
) -> bool:
    result = await session.execute(
        select(ResourcePermission)
        .where(
            ResourcePermission.resource_id == resource_id,
            ResourcePermission.user_id == user_id,
            ResourcePermission.operation == operation,
        )
    )
    return result.scalar_one_or_none() is not None

The immediate rationale is reasonable: the permission check requires a database query, and the repository is where database queries live. But the decision of whether access should be allowed is a business rule, not a persistence concern.

Once this pattern is established, it spreads. The repository accumulates check_* methods across multiple resources — agents, teams, documents, organisations — each encoding a fragment of the authorisation model. The authorisation logic is now distributed across multiple repository files, invisible to the application layer, and untestable without database access.

The DRY Rationalisation

A closely related form. Two services need to perform the same query with conditions that have business meaning. Rather than duplicating the query, one team moves it into the repository and calls it from both services.

# In a repository — but the filter conditions are a business decision
async def get_active_documents_for_user(
    self,
    session: AsyncSession,
    user_id: UUID,
    org_id: UUID,
) -> list[Document]:
    return await session.execute(
        select(Document)
        .where(
            Document.org_id == org_id,
            Document.owner_id == user_id,
            Document.status.in_(["published", "under_review"]),  # ← business rule
            Document.deleted_at.is_(None),                        # ← business rule
        )
    )

What counts as "active" — which statuses qualify, how deletion is defined — is a business decision. Moving it to the repository makes it a persistence decision. When the business definition of "active" changes, the change must be made in the data access layer, where no one expects to find it.

The Delayed Consequence

The repository pattern initially makes this feel contained. The repository is behind an interface, so the implementation seems swappable. But the interface now includes business semantics — check_user_access, get_active_documents_for_user — and those method names and signatures have encoded business rules into the persistence contract.

An in-memory adapter implementing the same interface for tests must also implement the same business logic. If the logic is complex, the in-memory and database implementations will drift apart over time. Tests pass; production fails.

The deeper problem: when a business rule changes, the change is invisible from the application layer. A developer reads await self._documents_repo.get_active_documents_for_user(...) and has no visibility into what "active" means, or where to change it.

Where the Logic Belongs

Access control belongs in the application layer, expressed in terms of domain rules, before the repository is called:

# Application layer — authorisation decision before persistence
class GetDocumentsHandler:
    async def handle(self, query: GetDocuments) -> list[Document]:
        actor = query.actor
 
        # Business rule: visible, testable without a database
        if not actor.has_permission(Permission.READ_DOCUMENTS):
            raise Forbidden("Insufficient permissions")
 
        return await self._document_repository.find_by_owner(
            owner_id=actor.user_id,
            org_id=actor.org_id,
        )

find_by_owner is a data access concern — it retrieves documents matching owner and org. The permission check is a business concern — it decides whether the caller is allowed to proceed. They belong in different layers.