Production Database Design with SQLAlchemy 2.0

In Phase 1 you set up the project skeleton. Now it is time to build the data foundation. Every production API needs a solid database layer -- and for TaskFlow, that means async SQLAlchemy models, Pydantic validation schemas, and version-controlled migrations with Alembic.

Why Async SQLAlchemy?

SQLAlchemy 2.0 introduced native async support through create_async_engine and AsyncSession. Combined with the asyncpg driver for PostgreSQL, this gives you non-blocking database access that matches FastAPI's async architecture.

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/taskflow"

engine = create_async_engine(DATABASE_URL, echo=False, pool_size=20)
async_session = async_sessionmaker(engine, expire_on_commit=False)

async def get_db():
    async with async_session() as session:
        yield session

Key points:

• create_async_engine replaces the synchronous create_engine
• async_sessionmaker (not the older sessionmaker) creates async-aware sessions
• expire_on_commit=False prevents lazy-load errors after commit in async contexts
• pool_size=20 is a sensible production starting point for connection pooling

TaskFlow Database Schema

TaskFlow needs four core tables with clear relationships:

Table	Purpose	Key Relationships
User	Accounts with auth data	Owns Projects, assigned Tasks
Project	Task containers	Has many Tasks, has Members
Task	Individual work items	Belongs to Project, assigned to User
ProjectMember	RBAC join table	Links User to Project with a role

The entity relationships look like this:

Two modelling decisions in that diagram are worth stating out loud, because they are the ones that get argued about in review:

assignee_id is nullable, project_id is not. A task with no assignee is a normal state — it is the backlog. A task with no project is not a state at all, it is corruption. Nullability is where you encode which of those two you believe.

The role lives on ProjectMember, not on User. Putting a role column on User would make roles global, and then "admin" would mean admin of everything. Because the role sits on the join row, the same person can own one project and be a read-only member of another — which is what people actually expect from a workspace tool.

Defining Models with SQLAlchemy 2.0

SQLAlchemy 2.0 uses DeclarativeBase and Mapped type annotations. This is a major shift from the legacy declarative_base() approach:

from datetime import datetime, timezone
from sqlalchemy import String, DateTime, ForeignKey, func, Enum as SAEnum
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
import enum

class Base(DeclarativeBase):
    pass

class TaskStatus(str, enum.Enum):
    todo = "todo"
    in_progress = "in_progress"
    done = "done"

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    full_name: Mapped[str] = mapped_column(String(100))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=lambda: datetime.now(timezone.utc),
    )

    # Relationships
    owned_projects: Mapped[list["Project"]] = relationship(back_populates="owner")
    assigned_tasks: Mapped[list["Task"]] = relationship(back_populates="assignee")
    memberships: Mapped[list["ProjectMember"]] = relationship(back_populates="user")

Two details in those timestamp columns are deliberate, and both are things that bite in production.

datetime.utcnow is not in this code, on purpose. It has been deprecated since Python 3.12, and the reason is worth understanding rather than memorising: it returns a naive datetime — one carrying no timezone — that happens to hold UTC. Every later comparison then has to remember a fact the object itself does not record, and eventually one of them forgets. The replacement is datetime.now(timezone.utc), which returns an aware datetime that cannot be misread. Note that it takes an argument, so as a column default it needs wrapping in a callable: lambda: datetime.now(timezone.utc).

server_default=func.now() beats a Python-side default for creation timestamps. It makes PostgreSQL stamp the row, so rows inserted by a migration, a seed script, or a psql session get a timestamp too — not just rows that happened to travel through your ORM. Pair it with DateTime(timezone=True) so the column is timestamptz and the database stores the offset instead of discarding it.

The other shift to notice is structural: Mapped[int] and mapped_column replace the old Column(Integer) style, which is what buys you type-checker support.

1from sqlalchemy import Column, Integer, String, Boolean
2from sqlalchemy.orm import declarative_base, relationship
3 
4Base = declarative_base()
5 
6class User(Base):
7    __tablename__ = "users"
8 
9    id = Column(Integer, primary_key=True)
10    email = Column(String(255), unique=True, index=True)
11    is_active = Column(Boolean, default=True)
12 
13    # your type checker sees: Column, not int
14    # user.id + 1  -> no warning, no help
15    projects = relationship("Project", back_populates="owner")

1from sqlalchemy import String
2from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
3 
4class Base(DeclarativeBase):
5    pass
6 
7class User(Base):
8    __tablename__ = "users"
9 
10    id: Mapped[int] = mapped_column(primary_key=True)
11    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
12    is_active: Mapped[bool] = mapped_column(default=True)
13 
14    # your type checker sees: int, str, bool
15    # nullability comes from Mapped[str] vs Mapped[str | None]
16    projects: Mapped[list["Project"]] = relationship(back_populates="owner")

The annotation is not decoration. In the 2.0 style, Mapped[str] and Mapped[str | None] are what declare a column NOT NULL or nullable — so the type checker and the database schema are derived from the same statement and cannot disagree. In the legacy style they were two separate declarations, and keeping them in sync was a manual job nobody ever did perfectly.

Why Alembic for Migrations

Alembic is the migration tool built for SQLAlchemy. Think of it as "git for your database schema" — with one important difference from git, which the widget below marks in amber.

Every migration is a Python file with upgrade() and downgrade() functions, which is what lets your team review schema changes in pull requests and keep environments in sync.

The amber step is the one to internalise. --autogenerate is a diff, not a mind reader: renaming full_name to display_name produces a drop_column followed by an add_column, which passes every test on an empty development database and silently deletes a production column's contents. Read the generated file every time, and rewrite that pair as op.alter_column(..., new_column_name=...) when what you meant was a rename.

Pydantic v2 Schemas

Pydantic v2 handles request/response validation. You create separate schemas for different operations:

from pydantic import BaseModel, EmailStr, ConfigDict
from datetime import datetime

class UserCreate(BaseModel):
    email: EmailStr
    password: str
    full_name: str

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

    id: int
    email: str
    full_name: str
    is_active: bool
    created_at: datetime

ConfigDict(from_attributes=True) replaces the old orm_mode = True from Pydantic v1. This lets you pass SQLAlchemy model instances directly to response schemas.

What You Will Build

In the lab that follows, you will:

1. Define all four SQLAlchemy 2.0 models with proper relationships and enums
2. Create an async database session factory
3. Write Pydantic v2 schemas for Create, Update, and Response variants of each model
4. Initialize Alembic and generate your first migration
5. Write a seed script to populate the database with test data

Next: Build the complete database layer in the hands-on lab. :::