Sqlalchemy Postgres

<essential_principles>

SQLAlchemy 2.0 + Pydantic + PostgreSQL Best Practices

This skill provides expert guidance for building production-ready database layers.

Stack

• SQLAlchemy 2.0 with async support (asyncpg driver)
• Pydantic v2 for validation and serialization
• Alembic for migrations
• PostgreSQL only

Core Principles

1. Separation of Concerns

models/       # SQLAlchemy ORM models (database layer)
schemas/      # Pydantic schemas (API layer)
repositories/ # Data access patterns
services/     # Business logic

2. Type Safety First Always use SQLAlchemy 2.0 style with Mapped[] type annotations:

from sqlalchemy.orm import Mapped, mapped_column

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

3. Async by Default Use async engine and sessions for FastAPI:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
engine = create_async_engine("postgresql+asyncpg://...")

4. Pydantic-SQLAlchemy Bridge Keep models and schemas separate but mappable:

# Schema reads from ORM
class UserRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

5. Repository Pattern Abstract database operations for testability and clean code. </essential_principles>

<intake> What do you need help with?

1. Setup database layer - Initialize SQLAlchemy + Pydantic + Alembic from scratch 2. Define models - Create SQLAlchemy models with Pydantic schemas 3. Create migration - Generate and manage Alembic migrations 4. Query patterns - Async CRUD, joins, eager loading, optimization 5. Full implementation - Complete database layer for a feature </intake>

<routing>

Auto-detection triggers (use this skill when user mentions):

• database, db, sqlalchemy, postgres, postgresql
• model, migration, alembic
• repository, crud, query
• async session, connection pool

</routing>

<reference_index>

Domain Knowledge

</reference_index>

<workflows_index>

</workflows_index>

<quick_reference>

File Structure

src/
├── db/
│   ├── __init__.py
│   ├── base.py          # DeclarativeBase
│   ├── session.py       # Engine + async session factory
│   └── dependencies.py  # FastAPI dependency
├── models/
│   ├── __init__.py
│   └── user.py          # SQLAlchemy models
├── schemas/
│   ├── __init__.py
│   └── user.py          # Pydantic schemas
├── repositories/
│   ├── __init__.py
│   ├── base.py          # Generic repository
│   └── user.py          # User repository
└── alembic/
    ├── alembic.ini
    ├── env.py
    └── versions/

Essential Imports

# Models
from sqlalchemy import String, Integer, ForeignKey, DateTime
from sqlalchemy.orm import Mapped, mapped_column, relationship, DeclarativeBase

# Async
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

# Pydantic
from pydantic import BaseModel, ConfigDict, Field

Connection String

# PostgreSQL async
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/dbname"

</quick_reference>

<success_criteria> Database layer is complete when:

• [ ] Async engine and session factory configured
• [ ] Base model with common fields (id, created_at, updated_at)
• [ ] Models use Mapped[] type annotations
• [ ] Pydantic schemas with from_attributes=True
• [ ] Alembic configured for async
• [ ] Repository pattern implemented
• [ ] FastAPI dependency for session injection
• [ ] Connection pooling configured for production

</success_criteria>

Async SQLAlchemy Patterns

<fastapi_integration>

FastAPI Integration

Session Dependency

from typing import Annotated, AsyncGenerator
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_factory() as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise

# Type alias for cleaner signatures
DBSession = Annotated[AsyncSession, Depends(get_db)]


# Usage in routes
@router.get("/users/{user_id}")
async def get_user(user_id: int, db: DBSession):
    user = await db.execute(select(User).where(User.id == user_id))
    return user.scalar_one_or_none()

Repository Dependency

def get_user_repo(session: DBSession) -> UserRepository:
    return UserRepository(session)

UserRepo = Annotated[UserRepository, Depends(get_user_repo)]


@router.get("/users")
async def list_users(repo: UserRepo):
    return await repo.get_multi()

Lifespan Event for Connection

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: verify connection
    async with engine.begin() as conn:
        await conn.execute(text("SELECT 1"))
    yield
    # Shutdown: dispose engine
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

</fastapi_integration>

<async_session_patterns>

Async Session Patterns

Context Manager Pattern

async def create_user(email: str) -> User:
    async with async_session_factory() as session:
        user = User(email=email)
        session.add(user)
        await session.commit()
        await session.refresh(user)
        return user

Manual Transaction Control

async def transfer_funds(from_id: int, to_id: int, amount: float):
    async with async_session_factory() as session:
        async with session.begin():
            # Both operations in same transaction
            from_account = await session.get(Account, from_id)
            to_account = await session.get(Account, to_id)

            if from_account.balance < amount:
                raise ValueError("Insufficient funds")

            from_account.balance -= amount
            to_account.balance += amount
            # Auto-commit on exit

Nested Transactions (Savepoints)

async def complex_operation():
    async with async_session_factory() as session:
        async with session.begin():
            session.add(user)

            try:
                async with session.begin_nested():
                    session.add(risky_operation)
            except Exception:
                # Only risky_operation rolled back
                pass

            # user still committed

</async_session_patterns>

<eager_loading_async>

Eager Loading in Async

Critical: Lazy loading doesn't work with async. Always use eager loading.

selectinload (Best for Collections)

# Loads related collection with SELECT ... IN (...)
stmt = select(User).options(selectinload(User.posts))
result = await session.execute(stmt)
users = result.scalars().all()

for user in users:
    print(user.posts)  # Already loaded, no additional query

joinedload (Best for Single Relations)

# Loads related object with JOIN
stmt = select(User).options(joinedload(User.organization))
result = await session.execute(stmt)
users = result.scalars().unique().all()  # Note: unique() needed with joins

Nested Eager Loading

stmt = select(User).options(
    selectinload(User.posts).selectinload(Post.comments),
    joinedload(User.organization),
)

contains_eager (With Explicit Join)

from sqlalchemy.orm import contains_eager

stmt = (
    select(User)
    .join(User.posts)
    .where(Post.is_published == True)
    .options(contains_eager(User.posts))
)

</eager_loading_async>

<concurrent_operations>

Concurrent Database Operations

Parallel Queries

import asyncio

async def get_dashboard_data(user_id: int):
    async with async_session_factory() as session:
        # Run queries in parallel
        user_task = session.execute(
            select(User).where(User.id == user_id)
        )
        posts_task = session.execute(
            select(Post).where(Post.author_id == user_id)
        )
        stats_task = session.execute(
            select(func.count(Post.id))
            .where(Post.author_id == user_id)
        )

        user_result, posts_result, stats_result = await asyncio.gather(
            user_task, posts_task, stats_task
        )

        return {
            "user": user_result.scalar_one(),
            "posts": posts_result.scalars().all(),
            "post_count": stats_result.scalar_one(),
        }

Batch Processing

async def process_users_batch(user_ids: list[int]):
    async with async_session_factory() as session:
        # Process in batches to avoid memory issues
        batch_size = 100

        for i in range(0, len(user_ids), batch_size):
            batch = user_ids[i:i + batch_size]

            result = await session.execute(
                select(User).where(User.id.in_(batch))
            )
            users = result.scalars().all()

            for user in users:
                await process_user(user)

            await session.commit()

</concurrent_operations>

<connection_pool>

Connection Pool Management

Configure for Production

engine = create_async_engine(
    DATABASE_URL,
    # Pool configuration
    pool_size=5,           # Maintain 5 connections
    max_overflow=10,       # Allow up to 15 total (5 + 10)
    pool_timeout=30,       # Wait 30s for available connection
    pool_recycle=1800,     # Recycle connections every 30 min
    pool_pre_ping=True,    # Check connection before use

    # Performance options
    echo=False,            # Disable SQL logging in production
    future=True,           # Use 2.0 style
)

Health Check Endpoint

@router.get("/health/db")
async def health_check(db: DBSession):
    try:
        await db.execute(text("SELECT 1"))
        return {"status": "healthy"}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

Graceful Shutdown

@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    # Properly close all connections
    await engine.dispose()

</connection_pool>

<error_handling>

Async Error Handling

Handle Database Errors

from sqlalchemy.exc import IntegrityError, OperationalError
from fastapi import HTTPException

async def create_user(user_data: UserCreate, db: DBSession):
    try:
        user = User(**user_data.model_dump())
        db.add(user)
        await db.commit()
        await db.refresh(user)
        return user

    except IntegrityError as e:
        await db.rollback()
        if "unique constraint" in str(e).lower():
            raise HTTPException(400, "Email already exists")
        raise HTTPException(400, "Database constraint violation")

    except OperationalError as e:
        await db.rollback()
        raise HTTPException(503, "Database unavailable")

Retry Pattern

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    reraise=True,
)
async def resilient_query(session: AsyncSession, user_id: int):
    result = await session.execute(
        select(User).where(User.id == user_id)
    )
    return result.scalar_one_or_none()

</error_handling>

<testing_async>

Testing Async Code

Pytest Fixture

import pytest
from httpx import AsyncClient, ASGITransport

@pytest.fixture
async def async_session():
    engine = create_async_engine(
        "postgresql+asyncpg://localhost/test_db",
        echo=False,
    )

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async with async_sessionmaker(engine)() as session:
        yield session

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

    await engine.dispose()


@pytest.fixture
async def client(async_session):
    async def override_get_db():
        yield async_session

    app.dependency_overrides[get_db] = override_get_db

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as client:
        yield client

    app.dependency_overrides.clear()


@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    response = await client.post(
        "/users/",
        json={"email": "test@test.com", "name": "Test"},
    )
    assert response.status_code == 201

</testing_async>

SQLAlchemy + Pydantic + PostgreSQL Best Practices

<security>

Security

Prevent SQL Injection

SQLAlchemy ORM and Core protect against SQL injection by default. Never use string formatting:

# NEVER DO THIS
query = f"SELECT * FROM users WHERE email = '{email}'"

# ALWAYS use parameterized queries
stmt = select(User).where(User.email == email)

# Or with text()
stmt = text("SELECT * FROM users WHERE email = :email")
result = await session.execute(stmt, {"email": email})

Protect Sensitive Data

# Store hashed passwords
from passlib.hash import bcrypt

class User(Base):
    password_hash: Mapped[str] = mapped_column(String(255))

    def set_password(self, password: str) -> None:
        self.password_hash = bcrypt.hash(password)

    def verify_password(self, password: str) -> bool:
        return bcrypt.verify(password, self.password_hash)

Environment Variables

Never hardcode credentials:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str

    class Config:
        env_file = ".env"

</security>

<performance>

Performance

Connection Pooling

Configure pool for production:

engine = create_async_engine(
    DATABASE_URL,
    pool_size=5,          # Base connections
    max_overflow=10,      # Extra connections under load
    pool_timeout=30,      # Wait time for connection
    pool_recycle=1800,    # Recycle connections every 30 min
    pool_pre_ping=True,   # Verify connection before use
)

Prevent N+1 Queries

Use eager loading:

# BAD - N+1 queries
users = await session.execute(select(User))
for user in users.scalars():
    print(user.posts)  # Triggers query for each user

# GOOD - Single query with selectinload
users = await session.execute(
    select(User).options(selectinload(User.posts))
)

Use Indexes

class User(Base):
    email: Mapped[str] = mapped_column(String(255), index=True)

    __table_args__ = (
        Index("ix_users_email_active", "email", "is_active"),
    )

Bulk Operations

# Bulk insert
await session.execute(
    insert(User),
    [{"email": "a@a.com"}, {"email": "b@b.com"}]
)

# Bulk update
await session.execute(
    update(User)
    .where(User.is_active == False)
    .values(status="inactive")
)

Select Only Needed Columns

# Instead of loading full objects
stmt = select(User.id, User.name)  # Faster for large tables

</performance>

<transactions>

Transaction Management

Use Context Manager

async with async_session_factory() as session:
    try:
        user = User(email="test@test.com")
        session.add(user)
        await session.commit()
    except Exception:
        await session.rollback()
        raise

Explicit Transactions

async with session.begin():
    # All operations in this block are atomic
    session.add(user)
    session.add(order)
    # Auto-commit on exit, auto-rollback on exception

Savepoints

async with session.begin():
    session.add(user1)

    async with session.begin_nested():
        # Savepoint
        session.add(user2)
        # Can rollback just this part

</transactions>

<testing>

Testing

Use Separate Test Database

@pytest.fixture
async def test_session():
    engine = create_async_engine(
        "postgresql+asyncpg://localhost/test_db",
        echo=False,
    )

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async with async_sessionmaker(engine)() as session:
        yield session

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

Transaction Rollback After Each Test

@pytest.fixture
async def session(test_engine):
    async with test_engine.connect() as conn:
        await conn.begin()  # Start transaction

        async_session = async_sessionmaker(bind=conn)
        async with async_session() as session:
            yield session

        await conn.rollback()  # Rollback after test

Factory Pattern for Test Data

from factory import Factory, Faker
from factory.alchemy import SQLAlchemyModelFactory

class UserFactory(SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session_persistence = "commit"

    email = Faker("email")
    name = Faker("name")
    is_active = True

</testing>

<pydantic_integration>

Pydantic Integration

Schema Naming Convention

{Entity}Base    - Shared fields
{Entity}Create  - Fields for creation
{Entity}Update  - Optional fields for updates
{Entity}Read    - Database output
{Entity}InDB    - Internal use with sensitive data

Validation

from pydantic import BaseModel, EmailStr, Field, field_validator

class UserCreate(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=1, max_length=100)
    age: int = Field(..., ge=0, le=150)

    @field_validator("name")
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Name cannot be empty")
        return v.strip()

Computed Fields

from pydantic import computed_field

class UserRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    first_name: str
    last_name: str

    @computed_field
    @property
    def full_name(self) -> str:
        return f"{self.first_name} {self.last_name}"

</pydantic_integration>

<common_mistakes>

Common Mistakes to Avoid

1. Forgetting `await` on async operations 2. Not closing sessions - Use context managers 3. Lazy loading in async - Causes errors, use eager loading 4. Mixing sync and async - Don't use sync driver with async code 5. N+1 queries - Always profile with echo=True during dev 6. Not using transactions - Wrap related operations 7. Hardcoding credentials - Use environment variables 8. Missing indexes - Profile slow queries 9. expire_on_commit=True with async - Set to False 10. Not handling IntegrityError - Catch and handle duplicates </common_mistakes>

Design Patterns for SQLAlchemy

<repository_pattern>

Repository Pattern

Abstracts data access logic from business logic.

from abc import ABC, abstractmethod
from typing import Generic, TypeVar, Optional, Sequence

T = TypeVar("T")


class AbstractRepository(ABC, Generic[T]):
    """Abstract base repository."""

    @abstractmethod
    async def get(self, id: int) -> Optional[T]:
        raise NotImplementedError

    @abstractmethod
    async def get_all(self) -> Sequence[T]:
        raise NotImplementedError

    @abstractmethod
    async def add(self, entity: T) -> T:
        raise NotImplementedError

    @abstractmethod
    async def delete(self, id: int) -> bool:
        raise NotImplementedError


class SQLAlchemyRepository(AbstractRepository[T]):
    """SQLAlchemy implementation of repository."""

    def __init__(self, model: type[T], session: AsyncSession):
        self.model = model
        self.session = session

    async def get(self, id: int) -> Optional[T]:
        result = await self.session.execute(
            select(self.model).where(self.model.id == id)
        )
        return result.scalar_one_or_none()

    async def get_all(self) -> Sequence[T]:
        result = await self.session.execute(select(self.model))
        return result.scalars().all()

    async def add(self, entity: T) -> T:
        self.session.add(entity)
        await self.session.flush()
        await self.session.refresh(entity)
        return entity

    async def delete(self, id: int) -> bool:
        result = await self.session.execute(
            delete(self.model).where(self.model.id == id)
        )
        return result.rowcount > 0

</repository_pattern>

<unit_of_work>

Unit of Work Pattern

Manages transactions across multiple repositories.

from types import TracebackType


class UnitOfWork:
    """Manages transaction lifecycle."""

    def __init__(self, session_factory: async_sessionmaker):
        self.session_factory = session_factory
        self.session: Optional[AsyncSession] = None

    async def __aenter__(self) -> "UnitOfWork":
        self.session = self.session_factory()
        self.users = UserRepository(self.session)
        self.orders = OrderRepository(self.session)
        return self

    async def __aexit__(
        self,
        exc_type: Optional[type[BaseException]],
        exc_val: Optional[BaseException],
        exc_tb: Optional[TracebackType],
    ) -> None:
        if exc_type is not None:
            await self.rollback()
        await self.session.close()

    async def commit(self) -> None:
        await self.session.commit()

    async def rollback(self) -> None:
        await self.session.rollback()


# Usage
async with UnitOfWork(session_factory) as uow:
    user = await uow.users.get(user_id)
    order = Order(user_id=user.id, total=100)
    await uow.orders.add(order)
    await uow.commit()

</unit_of_work>

<service_layer>

Service Layer Pattern

Business logic separate from data access.

class UserService:
    """Business logic for user operations."""

    def __init__(self, uow: UnitOfWork):
        self.uow = uow

    async def register_user(
        self,
        email: str,
        password: str,
        name: str,
    ) -> User:
        async with self.uow:
            # Check if user exists
            existing = await self.uow.users.get_by_email(email)
            if existing:
                raise ValueError("Email already registered")

            # Create user
            user = User(email=email, name=name)
            user.set_password(password)

            await self.uow.users.add(user)
            await self.uow.commit()

            return user

    async def deactivate_user(self, user_id: int) -> User:
        async with self.uow:
            user = await self.uow.users.get(user_id)
            if not user:
                raise ValueError("User not found")

            user.is_active = False
            await self.uow.commit()

            return user

</service_layer>

<soft_delete>

Soft Delete Pattern

Mark records as deleted instead of removing.

from datetime import datetime
from sqlalchemy import event


class SoftDeleteMixin:
    """Mixin for soft delete functionality."""

    deleted_at: Mapped[Optional[datetime]] = mapped_column(
        DateTime(timezone=True),
        nullable=True,
        default=None,
    )

    @property
    def is_deleted(self) -> bool:
        return self.deleted_at is not None

    def soft_delete(self) -> None:
        self.deleted_at = datetime.utcnow()

    def restore(self) -> None:
        self.deleted_at = None


class User(Base, SoftDeleteMixin):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255))


# Query only non-deleted
stmt = select(User).where(User.deleted_at.is_(None))

</soft_delete>

<audit_log>

Audit Log Pattern

Track changes to entities.

from sqlalchemy import event
import json


class AuditLog(Base):
    __tablename__ = "audit_logs"

    id: Mapped[int] = mapped_column(primary_key=True)
    table_name: Mapped[str] = mapped_column(String(100))
    record_id: Mapped[int] = mapped_column()
    action: Mapped[str] = mapped_column(String(10))  # INSERT, UPDATE, DELETE
    old_values: Mapped[Optional[str]] = mapped_column(Text)
    new_values: Mapped[Optional[str]] = mapped_column(Text)
    changed_by: Mapped[Optional[int]] = mapped_column()
    changed_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
    )


def audit_listener(mapper, connection, target, action):
    """Create audit log entry."""
    log = AuditLog(
        table_name=target.__tablename__,
        record_id=target.id,
        action=action,
        new_values=json.dumps(target.to_dict(), default=str),
    )
    connection.execute(insert(AuditLog).values(log.to_dict()))


# Register listeners
@event.listens_for(User, "after_insert")
def user_after_insert(mapper, connection, target):
    audit_listener(mapper, connection, target, "INSERT")


@event.listens_for(User, "after_update")
def user_after_update(mapper, connection, target):
    audit_listener(mapper, connection, target, "UPDATE")

</audit_log>

<specification_pattern>

Specification Pattern

Encapsulate query criteria.

from abc import ABC, abstractmethod
from sqlalchemy import Select


class Specification(ABC):
    """Base specification for filtering."""

    @abstractmethod
    def apply(self, query: Select) -> Select:
        raise NotImplementedError

    def __and__(self, other: "Specification") -> "AndSpecification":
        return AndSpecification(self, other)

    def __or__(self, other: "Specification") -> "OrSpecification":
        return OrSpecification(self, other)


class AndSpecification(Specification):
    def __init__(self, *specs: Specification):
        self.specs = specs

    def apply(self, query: Select) -> Select:
        for spec in self.specs:
            query = spec.apply(query)
        return query


# Concrete specifications
class ActiveUserSpec(Specification):
    def apply(self, query: Select) -> Select:
        return query.where(User.is_active == True)


class EmailDomainSpec(Specification):
    def __init__(self, domain: str):
        self.domain = domain

    def apply(self, query: Select) -> Select:
        return query.where(User.email.ilike(f"%@{self.domain}"))


# Usage
spec = ActiveUserSpec() & EmailDomainSpec("example.com")
stmt = spec.apply(select(User))

</specification_pattern>

<pagination>

Pagination Helper

from dataclasses import dataclass
from typing import Generic, TypeVar, Sequence

T = TypeVar("T")


@dataclass
class Page(Generic[T]):
    items: Sequence[T]
    total: int
    page: int
    per_page: int

    @property
    def pages(self) -> int:
        return (self.total + self.per_page - 1) // self.per_page

    @property
    def has_next(self) -> bool:
        return self.page < self.pages

    @property
    def has_prev(self) -> bool:
        return self.page > 1


async def paginate(
    session: AsyncSession,
    query: Select,
    page: int = 1,
    per_page: int = 20,
) -> Page:
    """Paginate a query."""
    # Count total
    count_query = select(func.count()).select_from(query.subquery())
    total = (await session.execute(count_query)).scalar_one()

    # Get items
    items_query = query.offset((page - 1) * per_page).limit(per_page)
    items = (await session.execute(items_query)).scalars().all()

    return Page(
        items=items,
        total=total,
        page=page,
        per_page=per_page,
    )


# Usage
page = await paginate(session, select(User), page=2, per_page=10)

</pagination>

Workflow: Create and Manage Migrations

<required_reading> Read before proceeding: 1. references/best-practices.md </required_reading>

<process>

Step 1: Generate Migration

After modifying models, generate a migration:

# Auto-generate migration from model changes
alembic revision --autogenerate -m "descriptive_message"

# Examples:
alembic revision --autogenerate -m "create_users_table"
alembic revision --autogenerate -m "add_email_to_users"
alembic revision --autogenerate -m "create_posts_and_comments"

Step 2: Review Generated Migration

ALWAYS review the generated migration file before applying!

Check alembic/versions/{revision}_descriptive_message.py:

"""create_users_table

Revision ID: abc123def456
Revises:
Create Date: 2024-01-15 10:30:00.000000

"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa

revision: str = 'abc123def456'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('email', sa.String(length=255), nullable=False),
        sa.Column('name', sa.String(length=100), nullable=True),
        sa.Column('created_at', sa.DateTime(timezone=True),
                  server_default=sa.text('now()'), nullable=False),
        sa.Column('updated_at', sa.DateTime(timezone=True),
                  server_default=sa.text('now()'), nullable=False),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('email'),
    )
    op.create_index('ix_users_email', 'users', ['email'], unique=True)


def downgrade() -> None:
    op.drop_index('ix_users_email', table_name='users')
    op.drop_table('users')

Step 3: Apply Migration

# Apply all pending migrations
alembic upgrade head

# Apply specific migration
alembic upgrade abc123def456

# Apply next migration only
alembic upgrade +1

Step 4: Common Migration Operations

Adding a Column

def upgrade() -> None:
    op.add_column('users', sa.Column('phone', sa.String(20), nullable=True))


def downgrade() -> None:
    op.drop_column('users', 'phone')

Adding Non-Nullable Column (with data)

def upgrade() -> None:
    # Add as nullable first
    op.add_column('users', sa.Column('status', sa.String(20), nullable=True))

    # Populate existing rows
    op.execute("UPDATE users SET status = 'active' WHERE status IS NULL")

    # Make non-nullable
    op.alter_column('users', 'status', nullable=False)


def downgrade() -> None:
    op.drop_column('users', 'status')

Adding Foreign Key

def upgrade() -> None:
    op.add_column('posts', sa.Column('author_id', sa.Integer(), nullable=True))
    op.create_foreign_key(
        'fk_posts_author_id',
        'posts', 'users',
        ['author_id'], ['id'],
        ondelete='CASCADE'
    )


def downgrade() -> None:
    op.drop_constraint('fk_posts_author_id', 'posts', type_='foreignkey')
    op.drop_column('posts', 'author_id')

Creating Index

def upgrade() -> None:
    op.create_index('ix_users_email_active', 'users', ['email', 'is_active'])


def downgrade() -> None:
    op.drop_index('ix_users_email_active', table_name='users')

Renaming Column

def upgrade() -> None:
    op.alter_column('users', 'name', new_column_name='full_name')


def downgrade() -> None:
    op.alter_column('users', 'full_name', new_column_name='name')

Creating Enum Type (PostgreSQL)

from sqlalchemy.dialects import postgresql

def upgrade() -> None:
    # Create enum type
    status_enum = postgresql.ENUM('pending', 'active', 'suspended', name='user_status')
    status_enum.create(op.get_bind())

    # Add column with enum
    op.add_column('users', sa.Column('status', status_enum, nullable=False,
                                     server_default='pending'))


def downgrade() -> None:
    op.drop_column('users', 'status')

    # Drop enum type
    status_enum = postgresql.ENUM('pending', 'active', 'suspended', name='user_status')
    status_enum.drop(op.get_bind())

Step 5: Migration Commands Reference

# Show current revision
alembic current

# Show migration history
alembic history

# Show pending migrations
alembic history --indicate-current

# Rollback last migration
alembic downgrade -1

# Rollback to specific revision
alembic downgrade abc123def456

# Rollback all migrations
alembic downgrade base

# Show SQL without executing
alembic upgrade head --sql

# Create empty migration (for manual edits)
alembic revision -m "manual_data_migration"

# Stamp database (mark as migrated without running)
alembic stamp head

Step 6: Data Migrations

For migrations that modify data:

from sqlalchemy.sql import table, column
from sqlalchemy import String, Integer

def upgrade() -> None:
    # Define table structure for data operations
    users = table('users',
        column('id', Integer),
        column('email', String),
        column('status', String),
    )

    # Update data
    op.execute(
        users.update()
        .where(users.c.status == 'inactive')
        .values(status='suspended')
    )


def downgrade() -> None:
    users = table('users',
        column('id', Integer),
        column('status', String),
    )

    op.execute(
        users.update()
        .where(users.c.status == 'suspended')
        .values(status='inactive')
    )

Step 7: Best Practices

1. One concern per migration - Don't mix schema changes with data migrations 2. Always test downgrade - Run alembic downgrade -1 then alembic upgrade head 3. Review autogenerated code - Alembic may miss some changes or generate incorrect code 4. Use descriptive names - add_phone_to_users not update_table 5. Keep migrations small - Easier to debug and rollback 6. Never modify applied migrations - Create a new migration instead 7. Handle NULL values - When adding non-nullable columns 8. Use transactions - Alembic wraps migrations in transactions by default </process>

<success_criteria> Migration is complete when:

• [ ] Migration generated with descriptive name
• [ ] upgrade() and downgrade() both reviewed
• [ ] Tested on local database
• [ ] downgrade tested (rollback works)
• [ ] No hardcoded values that differ between environments
• [ ] Data migrations handle edge cases
• [ ] Migration committed to version control

</success_criteria>

Workflow: Define Models and Schemas

<required_reading> Read before proceeding: 1. references/best-practices.md 2. references/patterns.md </required_reading>

<process>

Step 1: Understand the Entity

Before writing code, clarify:

• What data does this entity store?
• What are the relationships (one-to-many, many-to-many)?
• What fields are required vs optional?
• What fields need indexing?
• What constraints apply (unique, check)?

Step 2: Create SQLAlchemy Model

Create src/models/{entity}.py:

from datetime import datetime
from typing import TYPE_CHECKING, Optional
from uuid import UUID, uuid4

from sqlalchemy import String, Text, ForeignKey, Index, CheckConstraint
from sqlalchemy.dialects.postgresql import UUID as PG_UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship

from db.base import Base, TimestampMixin

if TYPE_CHECKING:
    from .related_model import RelatedModel


class User(Base, TimestampMixin):
    """User model with comprehensive field examples."""

    __tablename__ = "users"

    # Primary key options
    # Option 1: Auto-increment integer
    id: Mapped[int] = mapped_column(primary_key=True)

    # Option 2: UUID (recommended for distributed systems)
    # id: Mapped[UUID] = mapped_column(
    #     PG_UUID(as_uuid=True),
    #     primary_key=True,
    #     default=uuid4,
    # )

    # Required string field with constraints
    email: Mapped[str] = mapped_column(
        String(255),
        unique=True,
        nullable=False,
        index=True,
    )

    # Optional string field
    name: Mapped[Optional[str]] = mapped_column(String(100))

    # Text field for longer content
    bio: Mapped[Optional[str]] = mapped_column(Text)

    # Boolean with default
    is_active: Mapped[bool] = mapped_column(default=True)

    # Enum-like string with check constraint
    status: Mapped[str] = mapped_column(
        String(20),
        default="pending",
    )

    # Foreign key relationship
    organization_id: Mapped[Optional[int]] = mapped_column(
        ForeignKey("organizations.id", ondelete="SET NULL"),
    )

    # Relationships (lazy loading by default)
    organization: Mapped[Optional["Organization"]] = relationship(
        back_populates="users",
    )

    # One-to-many relationship
    posts: Mapped[list["Post"]] = relationship(
        back_populates="author",
        cascade="all, delete-orphan",
    )

    # Table-level constraints and indexes
    __table_args__ = (
        Index("ix_users_email_active", "email", "is_active"),
        CheckConstraint(
            "status IN ('pending', 'active', 'suspended')",
            name="ck_users_status",
        ),
    )

    def __repr__(self) -> str:
        return f"<User(id={self.id}, email={self.email})>"

Step 3: Create Pydantic Schemas

Create src/schemas/{entity}.py:

from datetime import datetime
from typing import Optional
from uuid import UUID

from pydantic import BaseModel, ConfigDict, EmailStr, Field


# Base schema with common fields
class UserBase(BaseModel):
    """Base schema with shared fields."""

    email: EmailStr
    name: Optional[str] = Field(None, max_length=100)
    bio: Optional[str] = None
    is_active: bool = True


# Schema for creating new records
class UserCreate(UserBase):
    """Schema for creating a user."""

    password: str = Field(..., min_length=8)
    organization_id: Optional[int] = None


# Schema for updating records
class UserUpdate(BaseModel):
    """Schema for updating a user. All fields optional."""

    email: Optional[EmailStr] = None
    name: Optional[str] = Field(None, max_length=100)
    bio: Optional[str] = None
    is_active: Optional[bool] = None
    organization_id: Optional[int] = None


# Schema for reading from database
class UserRead(UserBase):
    """Schema for reading a user from database."""

    model_config = ConfigDict(from_attributes=True)

    id: int
    status: str
    organization_id: Optional[int]
    created_at: datetime
    updated_at: datetime


# Schema with relationships
class UserWithPosts(UserRead):
    """User with related posts."""

    posts: list["PostRead"] = []


# Schema for list responses
class UserList(BaseModel):
    """Paginated list of users."""

    items: list[UserRead]
    total: int
    page: int
    per_page: int
    pages: int


# Avoid circular imports
from .post import PostRead

UserWithPosts.model_rebuild()

Step 4: Define Relationships

One-to-Many Example

# Parent model (Organization)
class Organization(Base, TimestampMixin):
    __tablename__ = "organizations"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

    # One organization has many users
    users: Mapped[list["User"]] = relationship(
        back_populates="organization",
        cascade="all, delete-orphan",
    )


# Child model (User)
class User(Base, TimestampMixin):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    organization_id: Mapped[Optional[int]] = mapped_column(
        ForeignKey("organizations.id", ondelete="CASCADE"),
    )

    organization: Mapped[Optional["Organization"]] = relationship(
        back_populates="users",
    )

Many-to-Many Example

from sqlalchemy import Table, Column, ForeignKey

# Association table
user_roles = Table(
    "user_roles",
    Base.metadata,
    Column("user_id", ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
    Column("role_id", ForeignKey("roles.id", ondelete="CASCADE"), primary_key=True),
)


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    roles: Mapped[list["Role"]] = relationship(
        secondary=user_roles,
        back_populates="users",
    )


class Role(Base):
    __tablename__ = "roles"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)
    users: Mapped[list["User"]] = relationship(
        secondary=user_roles,
        back_populates="roles",
    )

Many-to-Many with Extra Data

class UserRole(Base, TimestampMixin):
    """Association table with extra columns."""

    __tablename__ = "user_roles"

    user_id: Mapped[int] = mapped_column(
        ForeignKey("users.id", ondelete="CASCADE"),
        primary_key=True,
    )
    role_id: Mapped[int] = mapped_column(
        ForeignKey("roles.id", ondelete="CASCADE"),
        primary_key=True,
    )
    granted_by: Mapped[Optional[int]] = mapped_column(ForeignKey("users.id"))
    expires_at: Mapped[Optional[datetime]] = mapped_column()

    # Relationships
    user: Mapped["User"] = relationship(foreign_keys=[user_id])
    role: Mapped["Role"] = relationship()
    grantor: Mapped[Optional["User"]] = relationship(foreign_keys=[granted_by])

Step 5: Register Models

Update src/models/__init__.py:

from .user import User
from .organization import Organization
from .post import Post

__all__ = ["User", "Organization", "Post"]

Update src/schemas/__init__.py:

from .user import UserCreate, UserRead, UserUpdate, UserList, UserWithPosts
from .organization import OrganizationCreate, OrganizationRead

__all__ = [
    "UserCreate",
    "UserRead",
    "UserUpdate",
    "UserList",
    "UserWithPosts",
    "OrganizationCreate",
    "OrganizationRead",
]

Step 6: Import Models in Alembic

Update alembic/env.py to import all models:

# Import all models so Alembic can detect them
from models import User, Organization, Post

</process>

<success_criteria> Model definition is complete when:

• [ ] SQLAlchemy model uses Mapped[] type annotations
• [ ] All fields have appropriate types and constraints
• [ ] Relationships properly defined with back_populates
• [ ] Cascade rules set (especially for deletes)
• [ ] Indexes created for frequently queried columns
• [ ] Pydantic schemas created: Create, Read, Update
• [ ] Schemas use ConfigDict(from_attributes=True)
• [ ] Models imported in alembic/env.py
• [ ] Models exported in __init__.py

</success_criteria>

Workflow: Query Patterns and Repository

<required_reading> Read before proceeding: 1. references/patterns.md 2. references/async-patterns.md </required_reading>

<process>

Step 1: Create Base Repository

Create src/repositories/base.py:

from typing import Generic, TypeVar, Type, Optional, Sequence
from sqlalchemy import select, func, delete, update
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from db.base import Base

ModelType = TypeVar("ModelType", bound=Base)


class BaseRepository(Generic[ModelType]):
    """Generic repository with common CRUD operations."""

    def __init__(self, model: Type[ModelType], session: AsyncSession):
        self.model = model
        self.session = session

    async def get(self, id: int) -> Optional[ModelType]:
        """Get a single record by ID."""
        result = await self.session.execute(
            select(self.model).where(self.model.id == id)
        )
        return result.scalar_one_or_none()

    async def get_multi(
        self,
        *,
        skip: int = 0,
        limit: int = 100,
    ) -> Sequence[ModelType]:
        """Get multiple records with pagination."""
        result = await self.session.execute(
            select(self.model)
            .offset(skip)
            .limit(limit)
            .order_by(self.model.id)
        )
        return result.scalars().all()

    async def create(self, obj_in: dict) -> ModelType:
        """Create a new record."""
        db_obj = self.model(**obj_in)
        self.session.add(db_obj)
        await self.session.flush()
        await self.session.refresh(db_obj)
        return db_obj

    async def update(self, id: int, obj_in: dict) -> Optional[ModelType]:
        """Update a record by ID."""
        # Remove None values to avoid overwriting with NULL
        update_data = {k: v for k, v in obj_in.items() if v is not None}

        if not update_data:
            return await self.get(id)

        await self.session.execute(
            update(self.model)
            .where(self.model.id == id)
            .values(**update_data)
        )
        await self.session.flush()
        return await self.get(id)

    async def delete(self, id: int) -> bool:
        """Delete a record by ID."""
        result = await self.session.execute(
            delete(self.model).where(self.model.id == id)
        )
        await self.session.flush()
        return result.rowcount > 0

    async def count(self) -> int:
        """Count all records."""
        result = await self.session.execute(
            select(func.count()).select_from(self.model)
        )
        return result.scalar_one()

    async def exists(self, id: int) -> bool:
        """Check if record exists."""
        result = await self.session.execute(
            select(func.count())
            .select_from(self.model)
            .where(self.model.id == id)
        )
        return result.scalar_one() > 0

Step 2: Create Entity Repository

Create src/repositories/user.py:

from typing import Optional, Sequence
from sqlalchemy import select, or_
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload, joinedload

from models.user import User
from .base import BaseRepository


class UserRepository(BaseRepository[User]):
    """Repository for User operations."""

    def __init__(self, session: AsyncSession):
        super().__init__(User, session)

    async def get_by_email(self, email: str) -> Optional[User]:
        """Get user by email address."""
        result = await self.session.execute(
            select(User).where(User.email == email)
        )
        return result.scalar_one_or_none()

    async def get_active_users(
        self,
        *,
        skip: int = 0,
        limit: int = 100,
    ) -> Sequence[User]:
        """Get all active users."""
        result = await self.session.execute(
            select(User)
            .where(User.is_active == True)
            .offset(skip)
            .limit(limit)
            .order_by(User.created_at.desc())
        )
        return result.scalars().all()

    async def get_with_posts(self, id: int) -> Optional[User]:
        """Get user with eager-loaded posts."""
        result = await self.session.execute(
            select(User)
            .where(User.id == id)
            .options(selectinload(User.posts))
        )
        return result.scalar_one_or_none()

    async def search(
        self,
        query: str,
        *,
        skip: int = 0,
        limit: int = 100,
    ) -> Sequence[User]:
        """Search users by name or email."""
        search_term = f"%{query}%"
        result = await self.session.execute(
            select(User)
            .where(
                or_(
                    User.name.ilike(search_term),
                    User.email.ilike(search_term),
                )
            )
            .offset(skip)
            .limit(limit)
        )
        return result.scalars().all()

    async def get_by_organization(
        self,
        organization_id: int,
        *,
        skip: int = 0,
        limit: int = 100,
    ) -> Sequence[User]:
        """Get all users in an organization."""
        result = await self.session.execute(
            select(User)
            .where(User.organization_id == organization_id)
            .offset(skip)
            .limit(limit)
        )
        return result.scalars().all()

Step 3: Query Patterns

Basic Queries

from sqlalchemy import select, and_, or_, not_

# Simple select
stmt = select(User)
result = await session.execute(stmt)
users = result.scalars().all()

# Filter with where
stmt = select(User).where(User.is_active == True)

# Multiple conditions (AND)
stmt = select(User).where(
    and_(
        User.is_active == True,
        User.organization_id == org_id,
    )
)

# OR conditions
stmt = select(User).where(
    or_(
        User.name.ilike("%john%"),
        User.email.ilike("%john%"),
    )
)

# NOT
stmt = select(User).where(not_(User.is_active))

# IN clause
stmt = select(User).where(User.id.in_([1, 2, 3]))

# BETWEEN
stmt = select(User).where(User.created_at.between(start_date, end_date))

# IS NULL / IS NOT NULL
stmt = select(User).where(User.organization_id.is_(None))
stmt = select(User).where(User.organization_id.isnot(None))

Ordering and Pagination

from sqlalchemy import desc, asc

# Order by
stmt = select(User).order_by(User.created_at.desc())

# Multiple order by
stmt = select(User).order_by(User.name.asc(), User.id.desc())

# Pagination
stmt = select(User).offset(skip).limit(limit)

# Combined
stmt = (
    select(User)
    .where(User.is_active == True)
    .order_by(User.created_at.desc())
    .offset(skip)
    .limit(limit)
)

Eager Loading (Prevent N+1)

from sqlalchemy.orm import selectinload, joinedload, subqueryload

# selectinload - Separate SELECT IN query (best for collections)
stmt = select(User).options(selectinload(User.posts))

# joinedload - Single query with JOIN (best for single objects)
stmt = select(User).options(joinedload(User.organization))

# Nested eager loading
stmt = select(User).options(
    selectinload(User.posts).selectinload(Post.comments)
)

# Multiple relationships
stmt = select(User).options(
    selectinload(User.posts),
    joinedload(User.organization),
)

Joins

from sqlalchemy import join

# Implicit join (through relationship)
stmt = (
    select(User)
    .join(User.posts)
    .where(Post.title.ilike("%python%"))
)

# Explicit join
stmt = (
    select(User, Post)
    .join(Post, User.id == Post.author_id)
)

# Left outer join
stmt = (
    select(User, Post)
    .outerjoin(Post, User.id == Post.author_id)
)

# Select specific columns
stmt = select(User.id, User.name, Post.title).join(Post)

Aggregations

from sqlalchemy import func

# Count
stmt = select(func.count()).select_from(User)
result = await session.execute(stmt)
total = result.scalar_one()

# Count with filter
stmt = select(func.count()).select_from(User).where(User.is_active == True)

# Group by with aggregation
stmt = (
    select(User.organization_id, func.count(User.id).label("user_count"))
    .group_by(User.organization_id)
    .having(func.count(User.id) > 5)
)

# Multiple aggregations
stmt = select(
    func.count(User.id).label("total"),
    func.count(User.id).filter(User.is_active == True).label("active"),
)

Subqueries

# Subquery for filtering
active_org_subq = (
    select(Organization.id)
    .where(Organization.is_active == True)
    .scalar_subquery()
)

stmt = select(User).where(User.organization_id.in_(active_org_subq))

# Correlated subquery
post_count_subq = (
    select(func.count(Post.id))
    .where(Post.author_id == User.id)
    .correlate(User)
    .scalar_subquery()
)

stmt = select(User, post_count_subq.label("post_count"))

Step 4: Use in FastAPI Routes

from fastapi import APIRouter, Depends, HTTPException, Query
from typing import Annotated

from db.dependencies import DBSession
from repositories.user import UserRepository
from schemas.user import UserCreate, UserRead, UserUpdate, UserList

router = APIRouter(prefix="/users", tags=["users"])


def get_user_repo(session: DBSession) -> UserRepository:
    return UserRepository(session)


UserRepo = Annotated[UserRepository, Depends(get_user_repo)]


@router.get("/", response_model=UserList)
async def list_users(
    repo: UserRepo,
    page: int = Query(1, ge=1),
    per_page: int = Query(20, ge=1, le=100),
):
    skip = (page - 1) * per_page
    users = await repo.get_multi(skip=skip, limit=per_page)
    total = await repo.count()

    return UserList(
        items=users,
        total=total,
        page=page,
        per_page=per_page,
        pages=(total + per_page - 1) // per_page,
    )


@router.get("/{user_id}", response_model=UserRead)
async def get_user(user_id: int, repo: UserRepo):
    user = await repo.get(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user


@router.post("/", response_model=UserRead, status_code=201)
async def create_user(user_in: UserCreate, repo: UserRepo):
    existing = await repo.get_by_email(user_in.email)
    if existing:
        raise HTTPException(status_code=400, detail="Email already registered")
    return await repo.create(user_in.model_dump())


@router.patch("/{user_id}", response_model=UserRead)
async def update_user(user_id: int, user_in: UserUpdate, repo: UserRepo):
    user = await repo.update(user_id, user_in.model_dump(exclude_unset=True))
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user


@router.delete("/{user_id}", status_code=204)
async def delete_user(user_id: int, repo: UserRepo):
    deleted = await repo.delete(user_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="User not found")

</process>

<success_criteria> Repository is complete when:

• [ ] Base repository with generic CRUD operations
• [ ] Entity-specific repository with custom queries
• [ ] Eager loading used where appropriate
• [ ] Pagination implemented
• [ ] Search functionality added
• [ ] FastAPI routes use repository pattern
• [ ] Proper error handling (404, 400)
• [ ] Type hints throughout

</success_criteria>

Workflow: Setup Database Layer

<required_reading> Read before proceeding: 1. references/best-practices.md 2. references/async-patterns.md </required_reading>

<process>

Step 1: Install Dependencies

pip install sqlalchemy[asyncio] asyncpg alembic pydantic pydantic-settings

Or add to pyproject.toml/requirements.txt:

sqlalchemy[asyncio]>=2.0.0
asyncpg>=0.29.0
alembic>=1.13.0
pydantic>=2.0.0
pydantic-settings>=2.0.0

Step 2: Create Database Configuration

Create src/db/config.py:

from pydantic_settings import BaseSettings
from functools import lru_cache


class DatabaseSettings(BaseSettings):
    POSTGRES_USER: str
    POSTGRES_PASSWORD: str
    POSTGRES_HOST: str = "localhost"
    POSTGRES_PORT: int = 5432
    POSTGRES_DB: str

    # Connection pool settings
    POOL_SIZE: int = 5
    MAX_OVERFLOW: int = 10
    POOL_TIMEOUT: int = 30
    POOL_RECYCLE: int = 1800  # 30 minutes

    @property
    def async_database_url(self) -> str:
        return (
            f"postgresql+asyncpg://{self.POSTGRES_USER}:{self.POSTGRES_PASSWORD}"
            f"@{self.POSTGRES_HOST}:{self.POSTGRES_PORT}/{self.POSTGRES_DB}"
        )

    @property
    def sync_database_url(self) -> str:
        """For Alembic migrations"""
        return (
            f"postgresql://{self.POSTGRES_USER}:{self.POSTGRES_PASSWORD}"
            f"@{self.POSTGRES_HOST}:{self.POSTGRES_PORT}/{self.POSTGRES_DB}"
        )

    class Config:
        env_file = ".env"


@lru_cache
def get_db_settings() -> DatabaseSettings:
    return DatabaseSettings()

Step 3: Create Base Model

Create src/db/base.py:

from datetime import datetime
from typing import Any
from sqlalchemy import DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    """Base class for all models."""

    # Common type annotation map
    type_annotation_map = {
        datetime: DateTime(timezone=True),
    }

    def to_dict(self) -> dict[str, Any]:
        """Convert model to dictionary."""
        return {c.name: getattr(self, c.name) for c in self.__table__.columns}


class TimestampMixin:
    """Mixin for created_at and updated_at timestamps."""

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now(),
        nullable=False,
    )

Step 4: Create Async Session Factory

Create src/db/session.py:

from sqlalchemy.ext.asyncio import (
    create_async_engine,
    AsyncSession,
    async_sessionmaker,
    AsyncEngine,
)
from .config import get_db_settings


def create_engine() -> AsyncEngine:
    """Create async database engine with connection pooling."""
    settings = get_db_settings()

    return create_async_engine(
        settings.async_database_url,
        echo=False,  # Set True for SQL logging in dev
        pool_size=settings.POOL_SIZE,
        max_overflow=settings.MAX_OVERFLOW,
        pool_timeout=settings.POOL_TIMEOUT,
        pool_recycle=settings.POOL_RECYCLE,
        pool_pre_ping=True,  # Verify connections before use
    )


engine = create_engine()

async_session_factory = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autoflush=False,
)


async def get_session() -> AsyncSession:
    """Get async database session."""
    async with async_session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

Step 5: Create FastAPI Dependency

Create src/db/dependencies.py:

from typing import Annotated, AsyncGenerator
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from .session import async_session_factory


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """FastAPI dependency for database session."""
    async with async_session_factory() as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise


# Type alias for dependency injection
DBSession = Annotated[AsyncSession, Depends(get_db)]

Step 6: Initialize Alembic

cd src
alembic init alembic

Update alembic/env.py:

import asyncio
from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context

# Import your models
from db.base import Base
from db.config import get_db_settings
# Import all models here to register them
# from models.user import User

config = context.config
settings = get_db_settings()

# Set database URL
config.set_main_option("sqlalchemy.url", settings.sync_database_url)

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode."""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata)

    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    """Run migrations in 'online' mode with async engine."""
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()


def run_migrations_online() -> None:
    """Run migrations in 'online' mode."""
    asyncio.run(run_async_migrations())


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

Step 7: Create .env File

POSTGRES_USER=your_user
POSTGRES_PASSWORD=your_password
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
POSTGRES_DB=your_database

Step 8: Create Package Init Files

Create src/db/__init__.py:

from .base import Base, TimestampMixin
from .session import engine, async_session_factory, get_session
from .dependencies import get_db, DBSession
from .config import get_db_settings, DatabaseSettings

__all__ = [
    "Base",
    "TimestampMixin",
    "engine",
    "async_session_factory",
    "get_session",
    "get_db",
    "DBSession",
    "get_db_settings",
    "DatabaseSettings",
]

</process>

<success_criteria> Setup is complete when:

• [ ] All dependencies installed
• [ ] Database config with pydantic-settings
• [ ] Base model with TimestampMixin
• [ ] Async engine with connection pooling
• [ ] Session factory configured
• [ ] FastAPI dependency created
• [ ] Alembic initialized with async support
• [ ] .env file with database credentials
• [ ] Directory structure matches expected layout

</success_criteria>