DEV Community

A0mineTV
A0mineTV

Posted on

A Conventional FastAPI Architecture: From Zero to Production

#webdev #programming #backend #python

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
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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())
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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}
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

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"
Enter fullscreen mode Exit fullscreen mode

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}!")
Enter fullscreen mode Exit fullscreen mode

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}!")
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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"}
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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: {} }
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode
  • 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.”

Top comments (0)