Database Repository Pattern

Overview

Async SQL with protocol-based repositories. Domain models are plain dataclasses (not ORM entities). Infrastructure-layer implementations handle ORM mapping — domain code never sees it.

Repository Pattern

# Domain layer — pure protocol (in contracts)
class UserRepositoryProtocol(Protocol):
    async def find(self, user_id: str) -> Result[User, NotFoundError]: ...
    async def save(self, user: User) -> Result[User, DomainError]: ...
    async def delete(self, user_id: str) -> Result[None, DomainError]: ...

# Infrastructure layer — concrete implementation
class SqlUserRepository:
    def __init__(self, db: DatabaseProviderProtocol):
        self.db = db

    async def find(self, user_id: str) -> Result[User, NotFoundError]:
        async with self.db.scoped_context() as conn:
            row = await conn.fetch_row("SELECT * FROM users WHERE id = $1", user_id)
            if not row:
                return Err(UserNotFoundError(user_id))
            return Ok(User(id=row["id"], name=row["name"], email=row["email"]))

# DI binding
container.singleton(UserRepositoryProtocol, SqlUserRepository)

Domain Models

from lexigram.domain import DomainModel

@dataclass(frozen=True)
class User(DomainModel):
    id: str
    name: str
    email: str

DomainModel is a mixin providing serialization helpers; it applies @dataclass(init=False) automatically when a subclass lacks one (frozen only if you declare it). Not SQLAlchemy mapped objects — mapping lives in infrastructure only.

Generic Repository

from lexigram.sql import GenericRepository

class UserRepo(GenericRepository[User]):
    def __init__(self, provider: DatabaseProviderProtocol):
        super().__init__(
            provider,
            table_name="users",
            entity_class=User,
            key_field="id",              # default
            soft_delete_enabled=False,   # optional
        )

Migrations

lexigram db init                        # Create migrations/ directory
lexigram db migrate "add user table"    # Auto-generate migration
lexigram db upgrade                      # Apply pending migrations
lexigram db downgrade                    # Revert
lexigram db status / history             # Migration state

Migration File Structure

# migrations/0003_add_user_table.py
from lexigram.sql.migrations.base import Migration

class AddUserTable(Migration):
    def __init__(self):
        super().__init__(version="0003", description="Add user table")

    async def up(self, provider: DatabaseProviderProtocol) -> None:
        await provider.execute("""
            CREATE TABLE users (
                id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
                name TEXT NOT NULL,
                email TEXT UNIQUE NOT NULL,
                created_at TIMESTAMPTZ DEFAULT now()
            )
        """)

    async def down(self, provider: DatabaseProviderProtocol) -> None:
        await provider.execute("DROP TABLE IF EXISTS users")

SQLMigration(version, description, up_sql, down_sql) covers plain-SQL cases.

Config

The SQL config section is sql: (env prefix LEX_SQL__):

sql:
  backend:
    url: postgresql+asyncpg://user:pass@localhost:5432/mydb
  pool:
    min_size: 2
    max_size: 10
    max_overflow: 5
  operations:
    echo: false        # SQL logging

Use async drivers: postgresql+asyncpg://, mysql+aiomysql://, sqlite+aiosqlite://.

Testing

# Unit test with fake repo
class FakeUserRepo:
    def __init__(self):
        self._users: dict[str, User] = {}

    async def find(self, user_id: str) -> Result[User, NotFoundError]:
        user = self._users.get(user_id)
        if not user:
            return Err(UserNotFoundError(user_id))
        return Ok(user)

Common Mistakes

• Domain models extending SQLAlchemy Base — keeps ORM out of domain
• Business logic in repository — repositories query/store, services orchestrate
• Using sync drivers in async context — deadlock risk
• Not wrapping transactions — use scoped_context() or unit_of_work
• Mixing migration state across environments — always version-control migrations/