A Conventional FastAPI Architecture: From Zero to Production

This post outlines a conventional, production-friendly architecture for FastAPI. It trades cleverness for clarity: clear boundaries, typed models, async DB, JWT auth, migrations, tasks, metrics, tests, and containers. Copy the layout, paste the snippets, and ship.

TL;DR — The Folder Layout

fastapi-app/
├─ app/
│  ├─ main.py
│  ├─ core/
│  │  ├─ settings.py
│  │  ├─ logging.py
│  │  ├─ security.py
│  │  └─ deps.py
│  ├─ db/
│  │  ├─ base.py
│  │  ├─ session.py
│  │  └─ init.sql
│  ├─ models/
│  │  └─ user.py
│  ├─ schemas/
│  │  └─ user.py
│  ├─ api/
│  │  └─ v1/
│  │     ├─ router.py
│  │     ├─ endpoints/
│  │     │  ├─ auth.py
│  │     │  └─ users.py
│  │     └─ __init__.py
│  ├─ services/
│  │  └─ users.py
│  ├─ tasks/
│  │  ├─ celery_app.py
│  │  └─ jobs.py
│  ├─ telemetry/
│  │  └─ metrics.py
│  └─ __init__.py
├─ migrations/          # Alembic
│  ├─ env.py
│  └─ versions/
├─ tests/
│  ├─ conftest.py
│  └─ test_users.py
├─ .env.example
├─ alembic.ini
├─ docker-compose.yml
├─ Dockerfile
├─ Makefile
├─ pyproject.toml
└─ README.md

Why this layout? It creates seams: config/logging/security, DB plumbing, models/schemas, transport (API routers), business logic (services), async jobs (tasks), and ops (metrics, Docker, CI). Each piece can evolve without tangling the rest.

---

1) Goals & Principles

• Async-ready I/O (FastAPI + SQLAlchemy 2.0 async + asyncpg)
• Type safety (Pydantic v2 for schemas, modern typing)
• Migrations (Alembic) and reproducible environments
• Auth via JWT, with testable dependencies
• Observability (Prometheus metrics; easy to add tracing)
• Testability (PyTest + httpx)
• 12-Factor config with pydantic-settings
• Boring containerization (Docker/Compose) + Makefile

---

2) Core: settings, logging, security, dependencies

app/core/settings.py — one source of truth

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    APP_NAME: str = "FastAPI App"
    ENV: str = "local"
    DEBUG: bool = True

    SECRET_KEY: str = "change-me"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60
    JWT_ALG: str = "HS256"

    DATABASE_URL: str  # postgresql+asyncpg://user:pass@db:5432/app
    REDIS_URL: str = "redis://redis:6379/0"
    CORS_ORIGINS: list[str] = ["*"]

    model_config = SettingsConfigDict(env_file=".env", case_sensitive=False)

settings = Settings()

app/core/security.py — hashing + JWT

from datetime import datetime, timedelta, timezone
from jose import jwt
from passlib.context import CryptContext
from app.core.settings import settings

pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(p: str) -> str: return pwd_ctx.hash(p)
def verify_password(p: str, h: str) -> bool: return pwd_ctx.verify(p, h)

def create_access_token(sub: str, minutes: int | None = None) -> str:
    exp = datetime.now(timezone.utc) + timedelta(minutes=minutes or settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    return jwt.encode({"sub": sub, "exp": exp}, settings.SECRET_KEY, algorithm=settings.JWT_ALG)

app/core/deps.py — auth dependency (for protected routes)

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.settings import settings
from app.core.security import create_access_token
from app.db.session import get_session
from app.models.user import User
from jose import jwt

oauth2 = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

async def get_current_user(token: str = Depends(oauth2), session: AsyncSession = Depends(get_session)) -> User:
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.JWT_ALG])
        email = payload.get("sub")
        if not email:
            raise ValueError
    except Exception:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
    q = await session.execute(select(User).where(User.email == email))
    user = q.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
    return user

---

3) Database: async engine, session, base

app/db/session.py

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.core.settings import settings

engine = create_async_engine(settings.DATABASE_URL, pool_pre_ping=True, echo=settings.DEBUG)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False, autoflush=False, class_=AsyncSession)

async def get_session() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

4) Domain: models & schemas

app/models/user.py

from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String, Boolean, DateTime, func
from app.db.base import Base

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(320), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(128))
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[DateTime] = mapped_column(DateTime(timezone=True), server_default=func.now())

app/schemas/user.py (Pydantic v2)

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

class UserCreate(BaseModel):
    email: EmailStr
    password: str = Field(min_length=8)

class UserRead(BaseModel):
    id: int
    email: EmailStr
    is_active: bool
    created_at: datetime
    model_config = ConfigDict(from_attributes=True)

---

5) Transport: API routers (v1), endpoints

app/api/v1/endpoints/auth.py

from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, EmailStr
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.db.session import get_session
from app.models.user import User
from app.core.security import verify_password, create_access_token, hash_password

router = APIRouter(tags=["auth"])

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

class Login(BaseModel):
    email: EmailStr
    password: str

@router.post("/login", response_model=Token)
async def login(data: Login, session: AsyncSession = Depends(get_session)):
    user = (await session.execute(select(User).where(User.email == data.email))).scalar_one_or_none()
    if not user or not verify_password(data.password, user.hashed_password):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
    return Token(access_token=create_access_token(sub=user.email))

class Register(BaseModel):
    email: EmailStr
    password: str

@router.post("/register", status_code=201)
async def register(data: Register, session: AsyncSession = Depends(get_session)):
    if (await session.execute(select(User).where(User.email == data.email))).scalar_one_or_none():
        raise HTTPException(400, "Email already registered")
    user = User(email=data.email, hashed_password=hash_password(data.password))
    session.add(user)
    await session.commit()
    return {"id": user.id, "email": user.email}

app/api/v1/endpoints/users.py

from fastapi import APIRouter, Depends
from app.core.deps import get_current_user
from app.schemas.user import UserRead
from app.models.user import User

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

@router.get("/me", response_model=UserRead)
async def me(current: User = Depends(get_current_user)):
    return current

app/api/v1/router.py

from fastapi import APIRouter
from .endpoints import auth, users

api_router = APIRouter(prefix="/api/v1")
api_router.include_router(auth.router, prefix="/auth")
api_router.include_router(users.router)

---

6) Services: where business logic lives

Keep routers thin and move non-trivial work here.

app/services/users.py

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.models.user import User

async def get_user_by_email(session: AsyncSession, email: str) -> User | None:
    return (await session.execute(select(User).where(User.email == email))).scalar_one_or_none()

---

7) Tasks: background jobs (optional)

app/tasks/celery_app.py

from celery import Celery
from app.core.settings import settings
celery = Celery("app", broker=settings.REDIS_URL, backend=settings.REDIS_URL)
celery.conf.task_default_queue = "default"

app/tasks/jobs.py

from .celery_app import celery

@celery.task(name="send_welcome_email")
def send_welcome_email(email: str) -> None:
    print(f"Welcome, {email}!")

app/tasks/jobs.py

from .celery_app import celery

@celery.task(name="send_welcome_email")
def send_welcome_email(email: str) -> None:
    print(f"Welcome, {email}!")

---

8) Telemetry: metrics & health

app/telemetry/metrics.py

from fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator

def init_metrics(app: FastAPI) -> None:
    Instrumentator().instrument(app).expose(app, endpoint="/metrics", include_in_schema=False)

You’ll get Prometheus metrics at GET /metrics and a basic GET /health endpoint below.

---

9) App entrypoint: middlewares, routers, metrics

app/main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from app.core.settings import settings
from app.api.v1.router import api_router
from app.telemetry.metrics import init_metrics
from app.db.session import engine

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(lambda _: None)
    yield
    await engine.dispose()

app = FastAPI(title=settings.APP_NAME, lifespan=lifespan)

app.add_middleware(CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS, allow_credentials=True,
    allow_methods=["*"], allow_headers=["*"],
)
app.add_middleware(GZipMiddleware, minimum_size=1024)

init_metrics(app)

@app.get("/health", tags=["infra"])
async def health(): return {"status": "ok"}

app.include_router(api_router)

---

10) Migrations: Alembic (async-friendly)

• Configure alembic.ini and migrations/env.py to point at your metadata.
• First migration creates users table; run:

alembic revision -m "create users"
alembic upgrade head

Tip: keep migrations small and linear; review SQL diffs in PR.

---

11) Tests: PyTest + httpx

tests/conftest.py

import asyncio, pytest
from httpx import AsyncClient
from app.main import app

@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop
    loop.close()

@pytest.fixture
async def client():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

tests/test_users.py

import pytest

@pytest.mark.asyncio
async def test_health(client):
    r = await client.get("/health")
    assert r.status_code == 200
    assert r.json() == {"status": "ok"}

---

12) Tooling: pyproject, Makefile, Docker

• pyproject.toml pins FastAPI, SQLAlchemy 2.x, Pydantic v2, Alembic, etc.
• Makefile for dev ergonomics.
• Dockerfile (multi-stage optional) + docker-compose.yml for Postgres/Redis.

Makefile excerpt

dev:
    uvicorn app.main:app --reload
test:
    pytest -q
lint:
    ruff check .
fmt:
    ruff format .
migrate:
    alembic revision -m "$(m)"
upgrade:
    alembic upgrade head

Compose excerpt

services:
  api:
    build: .
    env_file: .env
    ports: ["8000:8000"]
    depends_on: [db, redis]
  db:
    image: postgres:16
    environment:
      POSTGRES_DB: app
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
    ports: ["5432:5432"]
    volumes: ["pgdata:/var/lib/postgresql/data"]
  redis:
    image: redis:7
    ports: ["6379:6379"]
volumes: { pgdata: {} }

---

13) How to run locally

cp .env.example .env
# Edit DATABASE_URL if needed
alembic upgrade head
uvicorn app.main:app --reload  # http://127.0.0.1:8000/docs

• Register → POST /api/v1/auth/register
• Login → POST /api/v1/auth/login (copy Bearer token)
• Me → GET /api/v1/users/me (Authorize with the token)
• Metrics → GET /metrics
• Health → GET /health

---

14) Patterns & Extensions

• Validation: thin endpoints + fat schemas (use Pydantic validators).
• Services: keep domain logic here; inject DB via Depends(get_session).
• Pagination: return X-Total-Count + Link headers.
• Security: add scopes/roles to JWT, rotate secrets, rate limit auth.
• Observability: add tracing via OpenTelemetry (FastAPI/SQLAlchemy/HTTPX instrumentors).
• Background jobs: Celery/Redis; for simpler needs, BackgroundTasks is often enough.
• CI: gate PRs with lint, type-check, tests; consider a smoke test container boot.
• Versioning: keep /api/v1 stable; put breaking changes in /api/v2.

---

15) Why this “conventional” structure works

• Discoverability: new devs find config, models, schemas, and API quickly.
• Testability: dependency-injected DB/session & pure services.
• Replaceability: swap Redis, add a queue, change DB driver—without surgery.
• Operational clarity: health, metrics, Docker—ops can run and observe it.

---

Final word

Architecture should be boring. This layout is: it lets you concentrate on domain work while keeping operations predictable. Fork it, tweak it, and share what you change—there’s always room for a better “boring default.”