Production REST APIs with FastAPI + PostgreSQL + Docker (2025)

This is the exact backend stack I use for automation-heavy projects — FastAPI, SQLAlchemy 2.0, Alembic, JWT auth, Celery for background tasks, and Docker Compose for local and production deployment. I've shipped this on AWS EC2 and ECS for clients in fintech, legaltech, and gym management.

Project Structure

app/
  api/
    routes/
      auth.py
      leads.py
      webhooks.py
    deps.py         # shared dependencies (DB session, current user)
  core/
    config.py       # pydantic settings from .env
    security.py     # JWT helpers
  db/
    base.py         # SQLAlchemy declarative base
    session.py      # async session factory
  models/           # SQLAlchemy ORM models
  schemas/          # Pydantic request/response schemas
  tasks/            # Celery task definitions
  main.py
docker-compose.yml
Dockerfile
alembic/
  env.py
  versions/

The Docker Compose Setup

version: "3.9"
services:
  api:
    build: .
    ports: ["8000:8000"]
    env_file: .env
    depends_on: [db, redis]
    volumes: ["./app:/app/app"]
    command: uvicorn app.main:app --host 0.0.0.0 --reload

  worker:
    build: .
    env_file: .env
    depends_on: [db, redis]
    command: celery -A app.tasks.celery_app worker --loglevel=info

  db:
    image: postgres:15
    environment:
      POSTGRES_USER: ${DB_USER}
      POSTGRES_PASSWORD: ${DB_PASS}
      POSTGRES_DB: ${DB_NAME}
    volumes: [postgres_data:/var/lib/postgresql/data]
    ports: ["5432:5432"]

  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]

volumes:
  postgres_data:

Async SQLAlchemy 2.0 Session

# db/session.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from app.core.config import settings

engine = create_async_engine(settings.DATABASE_URL, echo=False)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

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

JWT Authentication

# core/security.py
from datetime import datetime, timedelta
from jose import jwt, JWTError
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"])
SECRET_KEY = settings.SECRET_KEY
ALGORITHM = "HS256"

def create_access_token(subject: str, expires_delta: timedelta = timedelta(hours=24)):
    expire = datetime.utcnow() + expires_delta
    return jwt.encode({"sub": subject, "exp": expire}, SECRET_KEY, algorithm=ALGORITHM)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

A Real Route Example

# api/routes/leads.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.session import get_db
from app.api.deps import get_current_user
from app.models.lead import Lead
from app.schemas.lead import LeadCreate, LeadOut
from app.tasks.crm import process_lead_task

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

@router.post("/", response_model=LeadOut, status_code=201)
async def create_lead(
    payload: LeadCreate,
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_current_user)
):
    lead = Lead(**payload.model_dump(), created_by=current_user.id)
    db.add(lead)
    await db.commit()
    await db.refresh(lead)
    # Queue async processing (OpenAI classification, CRM sync)
    process_lead_task.delay(str(lead.id))
    return lead

Celery Background Task

# tasks/crm.py
from celery import Celery
from app.core.config import settings

celery_app = Celery("worker", broker=settings.REDIS_URL, backend=settings.REDIS_URL)

@celery_app.task(bind=True, max_retries=3)
def process_lead_task(self, lead_id: str):
    try:
        # 1. Fetch lead from DB (use sync session here)
        # 2. Call OpenAI to classify intent
        # 3. POST to Close CRM or HubSpot
        # 4. Send Slack notification
        pass
    except Exception as exc:
        raise self.retry(exc=exc, countdown=30)

Alembic Migrations

# Initialize
alembic init alembic

# Create a migration
alembic revision --autogenerate -m "create leads table"

# Apply
alembic upgrade head

The Dockerfile

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

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

This stack handles everything I need for automation backends — async DB operations, background task queuing, clean auth, and easy Docker deployment. I use it as the backend for n8n webhook receivers, AI agent APIs, and customer-facing portals.

Need a production-ready backend for your automation project? Let's talk.