database-patterns

Skill: Database Patterns

Database access patterns. Read when working with models, repositories, migrations.

---

Stack

• MariaDB 11+ (MySQL-compatible)
• SQLAlchemy 2.x async (asyncmy driver)
• Alembic for migrations
• Container: infra/docker-compose.yml -> service mariadb

Connection

• Env var: DATABASE_URL=mysql+asyncmy://${DB_USER}:${DB_PASSWORD}@mariadb:3306/${DB_NAME}
• Inside Docker use hostname mariadb (service name), NOT localhost.

Repository Pattern (Hexagonal)

The database is an adapter. Never import SQLAlchemy in the domain/application layer.

domain/
  ports.py          -> ABC interfaces (UserRepository, OrderRepository)
  entities.py       -> Pure Python classes (dataclass, no ORM)

application/
  services.py       -> Uses ports (DI), unaware of SQLAlchemy

infrastructure/
  adapters/
    db/
      models.py     -> SQLAlchemy ORM models (Base, User, Order)
      repositories.py -> Port implementations (SQLAlchemy queries)
      session.py    -> AsyncSession factory

Port (interface)

• File: domain/ports.py
• ABC class with @abstractmethod async methods (e.g. get_by_id, save, list_active)
• Return domain entities, not ORM models

Adapter (implementation)

• File: infrastructure/adapters/db/repositories.py
• Accept AsyncSession in __init__
• Use session.get() + .to_entity() for retrieval
• Use session.add() + session.flush() for persistence (NOT session.commit() - let session factory handle commit)

ORM Model

• File: infrastructure/adapters/db/models.py
• Inherit from DeclarativeBase
• Implement to_entity() instance method and from_entity() classmethod for domain entity conversion
• Use Column with explicit types, indexes, and constraints

Session Factory

• File: infrastructure/adapters/db/session.py
• create_async_engine + async_sessionmaker with expire_on_commit=False
• Wrap in async with session.begin(): to auto-commit/rollback
• Use as FastAPI dependency via Depends(get_session)