Building Scalable Backend APIs with FastAPI and PostgreSQL

Building scalable APIs isn't just about writing code that works—it's about architecting systems that can handle growth, maintain performance, and adapt to changing requirements. After building and scaling 15+ production APIs with FastAPI in 2025, I've learned that the framework you choose matters less than how you use it. But FastAPI with PostgreSQL? That's the combination that's consistently delivered the best results.

Let me share the exact architecture patterns, optimization techniques, and real performance data from APIs handling millions of requests daily.

Why FastAPI + PostgreSQL in 2025?

The FastAPI ecosystem has matured significantly. Combined with PostgreSQL's incredible feature set, it's become the go-to stack for serious backend development.

Framework Comparison (Dec 2025 Benchmarks)

FastAPI hits the sweet spot: nearly as fast as Go, but with Python's developer experience.

Architecture: The Foundation

Here's the architecture I use for every FastAPI project. It scales from MVP to millions of users.

Project Structure

backend/
├── app/
│   ├── __init__.py
│   ├── main.py                    # FastAPI app + middleware
│   ├── config.py                  # Pydantic settings
│   ├── database.py                # SQLAlchemy setup
│   ├── dependencies.py            # Dependency injection
│   │
│   ├── models/                    # SQLAlchemy models
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── product.py
│   │
│   ├── schemas/                   # Pydantic schemas
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── product.py
│   │
│   ├── api/                       # Route handlers
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── users.py
│   │   │   └── products.py
│   │
│   ├── services/                  # Business logic
│   │   ├── __init__.py
│   │   ├── user_service.py
│   │   └── product_service.py
│   │
│   ├── repositories/              # Data access
│   │   ├── __init__.py
│   │   ├── user_repo.py
│   │   └── product_repo.py
│   │
│   └── utils/                     # Helpers
│       ├── __init__.py
│       ├── cache.py
│       └── security.py
│
├── tests/
├── alembic/                       # Database migrations
├── requirements.txt
└── docker-compose.yml

This structure clearly separates concerns and scales beautifully.

The Complete Setup

1. Configuration with Pydantic Settings

# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    # Database
    DATABASE_URL: str
    DB_POOL_SIZE: int = 20
    DB_MAX_OVERFLOW: int = 10
    DB_POOL_TIMEOUT: int = 30

    # Redis Cache
    REDIS_URL: str
    CACHE_TTL: int = 3600

    # API
    API_TITLE: str = "My API"
    API_VERSION: str = "1.0.0"
    API_PREFIX: str = "/api/v1"

    # Security
    SECRET_KEY: str
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    ALGORITHM: str = "HS256"

    # Performance
    WORKERS: int = 4
    MAX_CONNECTIONS: int = 1000

    class Config:
        env_file = ".env"
        case_sensitive = True

@lru_cache()
def get_settings() -> Settings:
    return Settings()

2. Database Setup with Connection Pooling

# app/database.py
from sqlalchemy import create_engine, event
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import QueuePool
from app.config import get_settings

settings = get_settings()

# Optimized engine with connection pooling
engine = create_engine(
    settings.DATABASE_URL,
    poolclass=QueuePool,
    pool_size=settings.DB_POOL_SIZE,
    max_overflow=settings.DB_MAX_OVERFLOW,
    pool_timeout=settings.DB_POOL_TIMEOUT,
    pool_pre_ping=True,  # Verify connections before use
    echo=False,  # Set True for debugging
)

# Optimize PostgreSQL settings
@event.listens_for(engine, "connect")
def set_postgres_pragma(dbapi_conn, connection_record):
    cursor = dbapi_conn.cursor()
    cursor.execute("SET statement_timeout = '30s'")
    cursor.execute("SET idle_in_transaction_session_timeout = '60s'")
    cursor.close()

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Dependency for routes
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

3. Models with Optimized Indexes

# app/models/user.py
from sqlalchemy import Column, Integer, String, DateTime, Index, Boolean
from sqlalchemy.sql import func
from app.database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String(255), unique=True, nullable=False, index=True)
    username = Column(String(100), unique=True, nullable=False, index=True)
    hashed_password = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True)
    is_superuser = Column(Boolean, default=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    # Composite indexes for common queries
    __table_args__ = (
        Index('idx_email_active', 'email', 'is_active'),
        Index('idx_username_active', 'username', 'is_active'),
        Index('idx_created_at_desc', created_at.desc()),
    )

Repository Pattern for Data Access

# app/repositories/user_repo.py
from typing import Optional, List
from sqlalchemy.orm import Session
from sqlalchemy import select
from app.models.user import User
from app.schemas.user import UserCreate

class UserRepository:
    def __init__(self, db: Session):
        self.db = db

    def get_by_id(self, user_id: int) -> Optional[User]:
        return self.db.query(User).filter(User.id == user_id).first()

    def get_by_email(self, email: str) -> Optional[User]:
        # Uses index on email column
        return self.db.query(User)\
            .filter(User.email == email)\
            .first()

    def get_multi(
        self,
        skip: int = 0,
        limit: int = 100,
        active_only: bool = True
    ) -> List[User]:
        query = self.db.query(User)

        if active_only:
            query = query.filter(User.is_active == True)

        return query.offset(skip).limit(limit).all()

    def create(self, user_data: UserCreate) -> User:
        db_user = User(**user_data.dict())
        self.db.add(db_user)
        self.db.commit()
        self.db.refresh(db_user)
        return db_user

    def bulk_create(self, users: List[UserCreate]) -> List[User]:
        # Much faster than individual inserts
        db_users = [User(**user.dict()) for user in users]
        self.db.bulk_save_objects(db_users,  return_defaults=True)
        self.db.commit()
        return db_users

Service Layer with Business Logic

# app/services/user_service.py
from typing import Optional
from app.repositories.user_repo import UserRepository
from app.schemas.user import UserCreate, UserUpdate
from app.utils.security import get_password_hash
from app.utils.cache import cache_result

class UserService:
    def __init__(self, user_repo: UserRepository):
        self.user_repo = user_repo

    @cache_result(ttl=300)  # Cache for 5 minutes
    def get_user(self, user_id: int):
        return self.user_repo.get_by_id(user_id)

    async def create_user(self, user_data: UserCreate):
        # Hash password
        user_data.password = get_password_hash(user_data.password)

        # Create user
        user = self.user_repo.create(user_data)

        # Send welcome email (async task)
        await self.send_welcome_email(user.email)

        return user

    def validate_unique_email(self, email: str) -> bool:
        return self.user_repo.get_by_email(email) is None

API Routes with Proper Error Handling

# app/api/v1/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List

from app.database import get_db
from app.repositories.user_repo import UserRepository
from app.services.user_service import UserService
from app.schemas.user import User, UserCreate, UserUpdate

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

def get_user_service(db: Session = Depends(get_db)) -> UserService:
    user_repo = UserRepository(db)
    return UserService(user_repo)

@router.post("/", response_model=User, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_data: UserCreate,
    service: UserService = Depends(get_user_service)
):
    """
    Create a new user.

    - **email**: Must be unique
    - **password**: Minimum 8 characters
    """
    if not service.validate_unique_email(user_data.email):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Email already registered"
        )

    return await service.create_user(user_data)

@router.get("/{user_id}", response_model=User)
async def get_user(
    user_id: int,
    service: UserService = Depends(get_user_service)
):
    user = service.get_user(user_id)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )
    return user

@router.get("/", response_model=List[User])
async def list_users(
    skip: int = 0,
    limit: int = 100,
    service: UserService = Depends(get_user_service)
):
    return service.list_users(skip=skip, limit=limit)

Caching Strategy

Caching is critical for performance. Here's my implementation:

# app/utils/cache.py
import redis
import json
import functools
from app.config import get_settings

settings = get_settings()
redis_client = redis.from_url(settings.REDIS_URL)

def cache_result(ttl: int = 3600):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            # Generate cache key
            cache_key = f"{func.__name__}:{args}:{kwargs}"

            # Try to get from cache
            cached = redis_client.get(cache_key)
            if cached:
                return json.loads(cached)

            # Call function
            result = await func(*args, **kwargs)

            # Store in cache
            redis_client.setex(
                cache_key,
                ttl,
                json.dumps(result, default=str)
            )

            return result
        return wrapper
    return decorator

Caching Performance Impact

Database Optimization Techniques

1. Query Optimization

# ❌ Bad: N+1 Query Problem
users = db.query(User).all()
for user in users:
    posts = db.query(Post).filter(Post.user_id == user.id).all()

# ✅ Good: Eager Loading
from sqlalchemy.orm import joinedload

users = db.query(User)\
    .options(joinedload(User.posts))\
    .all()

2. Batch Operations

# ❌ Bad: Individual inserts
for user_data in user_list:
    user = User(**user_data)
    db.add(user)
    db.commit()

# ✅ Good: Bulk insert
users = [User(**data) for data in user_list]
db.bulk_save_objects(users)
db.commit()

Performance Comparison

Load Testing Results

I load-tested this architecture with Locust. Here are real numbers:

Test Configuration

• 10,000 concurrent users
• 5-minute test duration
• Mix of GET (70%), POST (20%), PUT/DELETE (10%)

Results

Monitoring and Observability

# app/main.py
from fastapi import FastAPI, Request
from prometheus_fastapi_instrumentator import Instrumentator
import time

app = FastAPI()

# Prometheus metrics
Instrumentator().instrument(app).expose(app)

# Request timing middleware
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

Production Deployment

Docker Compose Setup

version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis
    deploy:
      replicas: 4
      resources:
        limits:
          cpus: '0.5'
          memory: 512M

  db:
    image: postgres:16-alpine
    environment:
      - POSTGRES_DB=mydb
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
    volumes:
      - postgres_data:/var/lib/postgresql/data
    command:
      - "postgres"
      - "-c"
      - "max_connections=200"
      - "-c"
      - "shared_buffers=256MB"
      - "-c"
      - "effective_cache_size=1GB"

  redis:
    image: redis:7-alpine
    command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru

volumes:
  postgres_data:

Scaling Strategies

Horizontal Scaling Results

Sweet spot: 4 instances (best cost/performance ratio)

Security Best Practices

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

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

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

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

Conclusion

Building scalable APIs with FastAPI and PostgreSQL in 2025 is about following proven patterns, optimizing strategically, and monitoring religiously.

Key Takeaways

1. Architecture matters: Clear separation of concerns scales better
2. Cache aggressively: 90%+ faster response times
3. Optimize queries: Eager loading, indexing, batching
4. Monitor everything: Prometheus + Grafana for visibility
5. Test under load: Find bottlenecks before users do

This architecture has served me well across 15+ production APIs. Start with these patterns, measure everything, and optimize based on real data.

Building scalable APIs? Let's connect and share experiences. Drop your questions in the comments!