++++

Engineering

Apr 2026×12 min read

How the repository layer isolates SQLAlchemy from the rest of the application, translates between ORM entities and domain models, and keeps persistence concerns out of your business logic.

5. Repository Pattern

Driptanil DattaSoftware Developer

The Problem the Repository Solves

SQLAlchemy is powerful, but it's also opinionated. ORM-managed objects carry session state, lazy loading behavior, and identity map semantics. If you pass a TodoEntity (a SQLAlchemy model) directly to a service or a response schema, you're coupling everything above the database layer to SQLAlchemy's internal mechanics. Changing your ORM strategy, swapping databases, or writing unit tests without a real database all become significantly harder.

The repository pattern solves this by acting as a translation boundary. It takes Todo domain objects in, manages the persistence internally, and returns Todo domain objects out. Nothing above it ever sees a TodoEntity.

The Repository in Practice

TodoRepository in server/api/src/repositories/todo_repository.py accepts an AsyncSession at construction and exposes a set of explicit async methods: save, get_by_id, list_all, update, delete. Each one owns its own SQL, its own commit, and its own mapping.

Here is how save works:

# server/api/src/repositories/todo_repository.py
async def save(self, todo: Todo) -> None:
    entity = TodoEntity(
        id=str(todo.id),
        title=todo.title,
        description=todo.description,
        status=todo.status,
        created_at=todo.created_at,
        updated_at=todo.updated_at,
    )
    self.session.add(entity)
    await self.session.commit()

The domain Todo dataclass comes in. A TodoEntity SQLAlchemy model is constructed from it, added to the session, and committed. The service that called save never interacts with the entity — it just passes a domain object and trusts the repository to handle the rest.

Mapping Back to the Domain

The reverse mapping is equally explicit. _map_to_domain takes a TodoEntity and returns a clean Todo dataclass:

def _map_to_domain(self, entity: TodoEntity) -> Todo:
    return Todo(
        id=UUID(entity.id),
        title=entity.title,
        description=entity.description,
        status=entity.status,
        created_at=entity.created_at,
        updated_at=entity.updated_at,
    )

This is called internally by every read method. The get_by_id method, for example, runs the query, gets a nullable entity back, and converts it through _map_to_domain before returning:

async def get_by_id(self, todo_id: UUID) -> Todo | None:
    result = await self.session.execute(
        select(TodoEntity).where(TodoEntity.id == str(todo_id))
    )
    entity = result.scalar_one_or_none()
    if not entity:
        return None
    return self._map_to_domain(entity)

The UUID in, Todo | None out contract is what the service layer depends on. The query behind it is invisible to everything above.

What Leaking ORM Entities Looks Like

The common shortcut is to return the SQLAlchemy entity directly:

# What not to do
async def bad_get(self, todo_id: UUID) -> TodoEntity | None:
    return await self.session.get(TodoEntity, str(todo_id))

This pushes SQLAlchemy semantics into the service layer. The service now imports TodoEntity, knows about string-encoded UUIDs, and is entangled with session state. If the entity is detached from the session when the service tries to access a lazy-loaded attribute, you get a runtime error. If you want to test the service without a database, you can't — you'd need a real session to return a real entity.

Keeping Commit Boundaries Explicit

Each write method — save, update, delete — commits at the end of its own operation. This is intentional. It keeps transaction boundaries predictable and visible. You know exactly when data is persisted and why, because each method has a single, named purpose.

The update method shows this clearly:

async def update(self, todo: Todo) -> None:
    result = await self.session.execute(
        select(TodoEntity).where(TodoEntity.id == str(todo.id))
    )
    entity = result.scalar_one_or_none()
    if entity:
        entity.title = todo.title
        entity.description = todo.description
        entity.status = todo.status
        entity.updated_at = todo.updated_at
        await self.session.commit()

It fetches the existing entity, applies the changes from the domain object, and commits. The service never touches the entity. The entity never escapes the repository.

This structure means you can test TodoService.mark_completed by injecting a mock TodoRepository that returns a fake Todo object — no database, no session, no SQLAlchemy required.