Repository Pattern

Port-adapter repository pattern: domain interface, SQLAlchemy implementation, and domain-to-database mapping.

Show:

Status: Extracted
Purpose: Repository pattern implementation, domain-to-database mapping, and repository interfaces.

Repository Pattern Overview

Purpose

Encapsulates data access logic
Provides abstraction over data storage
Enables domain model independence from persistence
Supports testability through interfaces

Port-Adapter Pattern

Port (Interface)

Location: domain/*/repository.py or application/

Definition: Interface defined in domain/application layer that defines what data access is needed.

Pattern:

# Domain layer defines repository interface
class IAccountRepository(Protocol):
    """Account repository interface."""
    
    async def add(self, account: Account) -> None:
        """Add account to repository."""
        ...
    
    async def get_by_id(self, account_id: str) -> Account | None:
        """Get account by ID."""
        ...
    
    async def get_by_phone(self, phone: str) -> Account | None:
        """Get account by phone number."""
        ...

Adapter (Implementation)

Location: infrastructure/database/adapters/sqlalchemy/repositories/

Definition: Implementation of repository interface in infrastructure layer.

Pattern:

# Infrastructure layer implements repository
class SQLAlchemyAccountRepository:
    """SQLAlchemy implementation of account repository."""
    
    def __init__(self, session: AsyncSession):
        self.session = session
    
    async def add(self, account: Account) -> None:
        """Add account to repository."""
        account_model = self._to_model(account)
        self.session.add(account_model)
    
    async def get_by_id(self, account_id: str) -> Account | None:
        """Get account by ID."""
        result = await self.session.execute(
            select(AccountModel).where(AccountModel.id == account_id)
        )
        model = result.scalar_one_or_none()
        return self._to_model(account) if model else None

Domain-to-Database Mapping

Mapping Pattern

Domain Models:

Pure Python classes
No ORM annotations
Business logic and invariants
Location: domain/*/entities.py

Database Models:

SQLAlchemy ORM models
Database schema representation
Location: infrastructure/database/adapters/sqlalchemy/models/

Mappers:

Convert between domain and database models
Handle type conversions
Location: In repository or separate mapper files

Mapping Functions

Pattern:

def _to_domain(self, model: AccountModel) -> Account:
    """Map database model to domain entity."""
    return Account(
        id=model.id,
        phone=model.phone,
        created_at=model.created_at,
        # ... map all fields
    )
 
def _to_model(self, account: Account) -> AccountModel:
    """Map domain entity to database model."""
    return AccountModel(
        id=account.id,
        phone=account.phone,
        created_at=account.created_at,
        # ... map all fields
    )

Repository Responsibilities

Data Access

Query Execution: Execute database queries
Model Mapping: Convert between domain and database models
Session Management: Use provided session (from Unit of Work)

Domain Abstraction

Domain Interface: Repository interface defined in domain layer
Infrastructure Implementation: Implementation in infrastructure layer
No Domain Dependencies: Domain layer doesn't know about SQLAlchemy

Repository Integration with Unit of Work

Unit of Work Pattern

Pattern:

# Application layer uses Unit of Work
async def create_account(
    command: CreateAccountCommand,
    uow: IUnitOfWork,
) -> Account:
    """Create account use case."""
    async with uow:
        account = Account.create(...)
        uow.accounts.add(account)  # Repository accessed via Unit of Work
        # Transaction commits automatically
        return account

Benefits:

Repositories share same session
Transaction management handled by Unit of Work
Automatic commit/rollback

Repository Best Practices

Interface Design

Define in domain layer: Repository interfaces in domain
Keep interfaces focused: One repository per aggregate
Use async: All repository methods are async
Return domain objects: Always return domain entities, not database models

Implementation

Implement in infrastructure: Repository implementations in infrastructure
Handle mapping: Convert between domain and database models
Use session from Unit of Work: Don't create own sessions
Handle errors: Convert database errors to domain exceptions

Testing

Mock repositories: Easy to mock interfaces in tests
Test implementations: Integration tests for repository implementations
Test mapping: Verify domain-to-database mapping correctness

Repository Interface (Domain Layer)

Each aggregate has an abstract repository interface in domain/<feature>/repository.py. This is the port — no SQLAlchemy, no database, just Python ABCs.

# domain/account/repository.py
from abc import ABC, abstractmethod
from uuid import UUID
from domain.account.aggregate import Account
 
 
class AccountRepository(ABC):
    @abstractmethod
    async def save(self, session: object, account: Account) -> None: ...
 
    @abstractmethod
    async def find_by_id(self, session: object, account_id: UUID) -> Account | None: ...
 
    @abstractmethod
    async def find_by_email(self, session: object, email: str) -> Account | None: ...
 
    @abstractmethod
    async def delete(self, session: object, account_id: UUID) -> None: ...
 
    @abstractmethod
    async def find_all(
        self, session: object, limit: int = 50, offset: int = 0
    ) -> list[Account]: ...

session is typed as object here — the interface does not know whether the session is a SQLAlchemy AsyncSession or an in-memory dict. Concrete implementations know their session type.

Unit of Work Interface (Application Layer)

# application/unit_of_work.py
from abc import ABC, abstractmethod
from typing import Any
 
 
class UnitOfWork(ABC):
    @abstractmethod
    async def __aenter__(self) -> Any:
        """Returns the session object (type depends on adapter)."""
        ...
 
    @abstractmethod
    async def __aexit__(self, exc_type, exc, tb) -> None:
        """Commits on success, rolls back on exception."""
        ...

SQLAlchemy Implementations (Infrastructure Layer)

ORM Models (`infrastructure/database/adapters/sqlalchemy/models/account.py`)

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, DateTime
from uuid import UUID
import datetime
 
 
class Base(DeclarativeBase):
    pass
 
 
class AccountModel(Base):
    __tablename__ = "accounts"
 
    id: Mapped[UUID] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    display_name: Mapped[str] = mapped_column(String(255))
    status: Mapped[str] = mapped_column(String(50), default="active")
    created_at: Mapped[datetime.datetime] = mapped_column(DateTime(timezone=True))
    updated_at: Mapped[datetime.datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

Mappers (`infrastructure/database/adapters/sqlalchemy/mappers/account_mapper.py`)

from domain.account.aggregate import Account
from infrastructure.database.adapters.sqlalchemy.models.account import AccountModel
 
 
def account_to_orm(account: Account) -> AccountModel:
    return AccountModel(
        id=account.id,
        email=account.email,
        display_name=account.display_name,
        status=account.status.value,
        created_at=account.created_at,
        updated_at=account.updated_at,
    )
 
 
def account_to_domain(model: AccountModel) -> Account:
    return Account(
        id=model.id,
        email=model.email,
        display_name=model.display_name,
        status=AccountStatus(model.status),
        created_at=model.created_at,
        updated_at=model.updated_at,
    )

Repository Implementation (`infrastructure/database/adapters/sqlalchemy/repositories/account_repository.py`)

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from domain.account.aggregate import Account
from domain.account.repository import AccountRepository
from .mappers.account_mapper import account_to_orm, account_to_domain
 
 
class SQLAlchemyAccountRepository(AccountRepository):
    async def save(self, session: AsyncSession, account: Account) -> None:
        orm = account_to_orm(account)
        await session.merge(orm)  # insert or update
 
    async def find_by_id(self, session: AsyncSession, account_id) -> Account | None:
        result = await session.execute(
            select(AccountModel).where(AccountModel.id == account_id)
        )
        model = result.scalar_one_or_none()
        return account_to_domain(model) if model else None
 
    async def find_by_email(self, session: AsyncSession, email: str) -> Account | None:
        result = await session.execute(
            select(AccountModel).where(AccountModel.email == email)
        )
        model = result.scalar_one_or_none()
        return account_to_domain(model) if model else None
 
    async def delete(self, session: AsyncSession, account_id) -> None:
        model = await session.get(AccountModel, account_id)
        if model:
            await session.delete(model)
 
    async def find_all(self, session: AsyncSession, limit=50, offset=0) -> list[Account]:
        result = await session.execute(
            select(AccountModel).limit(limit).offset(offset)
        )
        return [account_to_domain(m) for m in result.scalars().all()]

UoW Implementation (`infrastructure/database/adapters/sqlalchemy/unit_of_work.py`)

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from application.unit_of_work import UnitOfWork
 
 
class SQLAlchemyUnitOfWork(UnitOfWork):
    def __init__(self, session_factory: async_sessionmaker):
        self._session_factory = session_factory
        self._session: AsyncSession | None = None
 
    async def __aenter__(self) -> AsyncSession:
        self._session = self._session_factory()
        return self._session
 
    async def __aexit__(self, exc_type, exc, tb) -> None:
        if exc:
            await self._session.rollback()
        else:
            await self._session.commit()
        await self._session.close()
        self._session = None

In-Memory Implementations (for Testing)

# infrastructure/database/adapters/inmemory/repositories/account_repository.py
from domain.account.repository import AccountRepository
from copy import deepcopy
 
 
class InMemoryAccountRepository(AccountRepository):
    async def save(self, session: dict, account: Account) -> None:
        session[str(account.id)] = deepcopy(account)
 
    async def find_by_id(self, session: dict, account_id) -> Account | None:
        return deepcopy(session.get(str(account_id)))
 
    async def find_by_email(self, session: dict, email: str) -> Account | None:
        return next(
            (deepcopy(a) for a in session.values() if a.email == email), None
        )
 
    async def delete(self, session: dict, account_id) -> None:
        session.pop(str(account_id), None)
 
    async def find_all(self, session: dict, limit=50, offset=0) -> list[Account]:
        return [deepcopy(a) for a in list(session.values())[offset:offset + limit]]

# infrastructure/database/adapters/inmemory/unit_of_work.py
from application.unit_of_work import UnitOfWork
 
 
class InMemoryUnitOfWork(UnitOfWork):
    def __init__(self, store: dict):
        self._store = store
 
    async def __aenter__(self) -> dict:
        return self._store  # no transactions — simple dict
 
    async def __aexit__(self, exc_type, exc, tb) -> None:
        pass  # nothing to commit or rollback

Factory (`infrastructure/database/factories.py`)

from config.settings import Settings
from application.unit_of_work import UnitOfWork
 
ADAPTER_REGISTRY = {
    "sqlalchemy": _build_sqlalchemy_uow,
    "inmemory": _build_inmemory_uow,
}
 
def build_unit_of_work(settings: Settings) -> UnitOfWork:
    builder = ADAPTER_REGISTRY[settings.database_adapter]
    return builder(settings)

Usage in Application Handlers

# application/account/create.py
class CreateAccountHandler:
    def __init__(self, account_repo: AccountRepository, unit_of_work: UnitOfWork):
        self._repo = account_repo
        self._uow = unit_of_work
 
    async def execute(self, command: CreateAccountCommand) -> Account:
        async with self._uow as session:
            existing = await self._repo.find_by_email(session, command.email)
            if existing:
                raise ConflictError("Email already in use")
 
            account = Account.create(
                email=command.email,
                display_name=command.display_name,
            )
            await self._repo.save(session, account)
            return account
        # UoW commits here on __aexit__ with no exception

Rules

Repositories persist and retrieve aggregates only — never sub-entities directly.
The session parameter is always obtained from the UoW — never created inside a repository.
Mappers (account_to_orm, account_to_domain) are standalone functions, not methods on the model or aggregate.
The in-memory adapter must have feature parity with the SQLAlchemy adapter — every test that passes with in-memory must be runnable against SQLAlchemy.
Migrations live in infrastructure/database/migrations/ (Alembic). See 06-database-persistence/01-migration-patterns.md.

Actor Context Authentication Patterns

Repository Pattern