Initial setup: FastAPI + SQLAlchemy + Alembic

You start by building the base. Folder structure, dependencies, Docker configuration, and a first endpoint that starts up. By the end of this capsule you have a project that starts with docker-compose up, connects to PostgreSQL, and answers a health-check GET. It's the base we'll add patterns onto in the following capsules.

---

Project structure

Create the directory:

mkdir taskflow
cd taskflow
git init

Initial structure:

taskflow/
├── README.md
├── docker-compose.yml
├── Dockerfile
├── pyproject.toml
├── .gitignore
├── alembic.ini
├── alembic/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── database.py
│   └── config.py
└── tests/
    ├── __init__.py
    └── conftest.py

---

pyproject.toml

[project]
name = "taskflow"
version = "0.1.0"
requires-python = ">=3.12"

dependencies = [
    "fastapi==0.110.0",
    "uvicorn[standard]==0.27.0",
    "sqlalchemy[asyncio]==2.0.25",
    "asyncpg==0.29.0",
    "alembic==1.13.1",
    "pydantic==2.5.3",
    "pydantic-settings==2.1.0",
    "pyjwt==2.10.1",  # JWT mock
    "python-multipart==0.0.6",
]

[project.optional-dependencies]
dev = [
    "pytest==7.4.4",
    "pytest-asyncio==0.23.3",
    "httpx==0.26.0",
    "testcontainers==3.7.1",
]

[tool.pytest.ini_options]
asyncio_mode = "auto"

Install:

python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

---

docker-compose.yml

version: '3.8'

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: taskflow
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 5s
      retries: 5

  pgbouncer:
    image: edoburu/pgbouncer:latest
    environment:
      DB_HOST: postgres
      DB_USER: postgres
      DB_PASSWORD: postgres
      DB_NAME: taskflow
      POOL_MODE: transaction
      MAX_CLIENT_CONN: 200
      DEFAULT_POOL_SIZE: 20
    ports:
      - "6432:5432"
    depends_on:
      postgres:
        condition: service_healthy

  app:
    build: .
    environment:
      DATABASE_URL: postgresql+asyncpg://postgres:postgres@pgbouncer:5432/taskflow
      JWT_SECRET: dev-secret-change-in-prod
    ports:
      - "8000:8000"
    depends_on:
      - pgbouncer
    volumes:
      - ./app:/app/app
      - ./alembic:/app/alembic
      - ./alembic.ini:/app/alembic.ini

volumes:
  postgres_data:

---

Dockerfile

FROM python:3.12-slim

WORKDIR /app

COPY pyproject.toml .
RUN pip install --no-cache-dir -e .

COPY . .

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]

---

app/config.py

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    database_url: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/taskflow"
    jwt_secret: str = "dev-secret"
    jwt_algorithm: str = "HS256"

    class Config:
        env_file = ".env"


settings = Settings()

---

app/database.py

from sqlalchemy.ext.asyncio import (
    create_async_engine,
    async_sessionmaker,
    AsyncSession,
)
from sqlalchemy.orm import DeclarativeBase

from app.config import settings


engine = create_async_engine(
    settings.database_url,
    pool_size=20,
    max_overflow=20,
    pool_pre_ping=True,
    pool_recycle=300,
    # CRITICAL: for PgBouncer transaction mode
    connect_args={"statement_cache_size": 0},
)

SessionLocal = async_sessionmaker(engine, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


async def get_db() -> AsyncSession:
    """Dependency for FastAPI."""
    async with SessionLocal() as session:
        yield session

statement_cache_size=0 is the gotcha from module 6 of guide #12 — without it, prepared statements break randomly with PgBouncer transaction mode.

---

app/main.py

from fastapi import FastAPI

app = FastAPI(
    title="TaskFlow API",
    description="Multi-tenant task management API with production patterns",
    version="0.1.0",
)


@app.get("/health")
async def health():
    return {"status": "ok"}

---

Configure Alembic

alembic init -t async alembic

This creates:

• alembic.ini: main config.
• alembic/env.py: async setup.
• alembic/script.py.mako: template for migrations.
• alembic/versions/: directory for migrations.

Edit alembic/env.py to use our config:

# alembic/env.py
from logging.config import fileConfig
import asyncio

from alembic import context
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import async_engine_from_config

from app.config import settings
from app.database import Base
# Import all models so Alembic discovers them
# from app.models import *  # (we'll add this later)


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

# Set database_url from settings
config.set_main_option("sqlalchemy.url", settings.database_url)

target_metadata = Base.metadata


def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")
    context.configure(url=url, target_metadata=target_metadata, literal_binds=True)
    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations():
    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():
    asyncio.run(run_async_migrations())


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

---

.gitignore

.venv/
__pycache__/
*.pyc
.env
.pytest_cache/
.DS_Store
postgres_data/

---

Initial README.md

# TaskFlow API

Multi-tenant task management API demonstrating production patterns:
- Cursor pagination
- Soft delete
- Audit logs (PostgreSQL triggers)
- Multi-tenancy (Row-Level Security)
- Zero-downtime migrations
- Optimistic locking
- Bulk operations (COPY + ON CONFLICT)

## Setup

```bash
docker-compose up -d
alembic upgrade head
curl http://localhost:8000/health

Run tests

pip install -e ".[dev]"
pytest

Docs

• BENCHMARKS.md: performance benchmarks
• MULTITENANCY.md: multi-tenancy architecture decisions
• RUNBOOK-MIGRATION.md: zero-downtime migration runbook

---

## Test the setup

```bash
# Bring it up
docker-compose up -d

# View logs
docker-compose logs -f app

# Test health endpoint
curl http://localhost:8000/health
# {"status": "ok"}

# Test DB connection
docker-compose exec postgres psql -U postgres -d taskflow -c "SELECT 1"
# Should return 1

If everything works, the base setup is ready. Commit:

git add .
git commit -m "feat: initial setup with FastAPI + SQLAlchemy + Alembic + PgBouncer"

---

Create the first table: tenants

As a teaser for the next capsule (multi-tenancy), we create the tenants table:

# app/models/__init__.py
from app.models.tenant import Tenant

__all__ = ["Tenant"]

# app/models/tenant.py
from datetime import datetime, timezone
import uuid
from sqlalchemy import String, DateTime
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class Tenant(Base):
    __tablename__ = "tenants"

    id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        primary_key=True,
        default=uuid.uuid4,
    )
    name: Mapped[str] = mapped_column(String(200), unique=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
    )

Create the first migration:

alembic revision --autogenerate -m "initial: create tenants table"

Check the generated file at alembic/versions/001_initial_create_tenants_table.py:

def upgrade() -> None:
    op.create_table(
        'tenants',
        sa.Column('id', sa.UUID(), nullable=False),
        sa.Column('name', sa.String(200), nullable=False),
        sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('name'),
    )


def downgrade() -> None:
    op.drop_table('tenants')

Apply:

alembic upgrade head

Verify:

docker-compose exec postgres psql -U postgres -d taskflow -c "\dt"
# Should list the 'tenants' table

Commit:

git add .
git commit -m "feat: add tenants table with first migration"

---

Final structure after this capsule

taskflow/
├── README.md
├── docker-compose.yml
├── Dockerfile
├── pyproject.toml
├── .gitignore
├── alembic.ini
├── alembic/
│   ├── env.py
│   ├── script.py.mako
│   └── versions/
│       └── 001_initial_create_tenants_table.py
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── database.py
│   ├── config.py
│   └── models/
│       ├── __init__.py
│       └── tenant.py
└── tests/
    ├── __init__.py
    └── conftest.py

---

Pitfalls and common mistakes

1. Forgetting statement_cache_size=0 with PgBouncer.

PgBouncer transaction mode breaks asyncpg's prepared statements. Without that config, random InvalidSQLStatementNameError errors. See module 6 of guide #12.

2. Mixing SQLAlchemy versions.

This project uses SQLAlchemy 2.0 style (mappers with Mapped[]). If you find tutorials with Column(), relationship() with a string, etc., those are 1.x style — convert them.

3. alembic.ini not updated.

alembic init -t async alembic creates alembic.ini with default values. The config comes from env.py, which reads settings.database_url. If you want to run Alembic standalone (not via the app), set sqlalchemy.url in alembic.ini.

4. Migrations with --autogenerate without importing the models.

Alembic detects changes by comparing metadata vs the DB. If you don't import your models in env.py, autogenerate detects nothing. Import all the models at the start of env.py:

from app.models import *  # noqa

5. Healthcheck in docker-compose without waiting for postgres to be ready.

Without condition: service_healthy, the app can start before postgres is ready and crash. Use the healthcheck.

6. Adding Mapped[] without importing sqlalchemy.orm.Mapped.

# ❌ Error
from sqlalchemy import Mapped  # doesn't exist

# ✅
from sqlalchemy.orm import Mapped, mapped_column

---

Summary and next step

What you have now:

• The project structure ready (FastAPI + SQLAlchemy 2.0 async + Alembic + PgBouncer).
• Docker compose that starts everything.
• The tenants table created with the first migration.
• The /health endpoint that verifies it starts.
• A repo on GitHub (after git push) with atomic commits.

Before moving on, you should:

• Verify docker-compose up starts without errors.
• Verify curl localhost:8000/health returns {"status": "ok"}.
• Verify alembic upgrade head applies the migration.
• Do a git push to the public repo.

In the next capsule we add what defines TaskFlow as a SaaS: multi-tenancy with Row-Level Security + mock auth. You'll create the users table and add tenant_id to all of them. Configure RLS policies. Implement a FastAPI dependency that sets app.tenant_id before every query. And the first aggressive isolation test: tenant A trying to read tenant B's data.

---

Resources

1. FastAPI — Tutorial — reference.
2. SQLAlchemy 2.0 — ORM Mapped Style — modern syntax.
3. Alembic — Async — async setup.
4. PgBouncer — Configuration — reference.
5. Pydantic Settings — configuration management.
6. Docker Compose — Healthchecks — reference.

---

Capsule 02 of 08 — Module 8 — SQL Patterns for Production APIs Guide