FastAPI + PostgreSQL + Redis: A Practical Stack

FastAPI, PostgreSQL, and Redis show up together in a lot of Python APIs, and for a practical reason. FastAPI handles the HTTP surface and validation. PostgreSQL holds the durable, relational data. Redis sits in front of expensive reads as a cache, so you are not hitting the database for the same item on every request.

This article wires those three together in a way you can actually run: SQLAlchemy 2.0 with asyncpg, connection pools created in a lifespan hook, Alembic for migrations, a cache-aside pattern with invalidation on write, and Docker Compose for local Postgres and Redis. It is a working baseline, not a claim about how much traffic the stack can take.

What each part should own

• FastAPI is the entrypoint: routes, request and response models, and dependencies. It should not open a new database engine on every request.
• PostgreSQL, through SQLAlchemy 2.0 async and Alembic, is the source of truth. If the cache and the database disagree, the database wins.
• Redis stores short-lived copies of read results. It is not the system of record. Rate limiting and sessions are valid later uses of the same server; they are not required for this baseline.

Keep configuration in the environment. A settings object with database_url and redis_url is enough to start. The database URL must use the postgresql+asyncpg:// scheme so SQLAlchemy selects the async driver.

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    database_url: str
    redis_url: str = "redis://localhost:6379/0"
    cache_ttl_seconds: int = 60

Run PostgreSQL and Redis with Docker Compose

Do not assume Postgres and Redis are already installed on a laptop. Compose gives you both, with health checks so the API does not start before they accept connections. The passwords below are for local development only.

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: postgresql+asyncpg://app:app@db:5432/app
      REDIS_URL: redis://redis:6379/0
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_healthy

  db:
    image: postgres:17
    environment:
      POSTGRES_USER: app
      POSTGRES_PASSWORD: app
      POSTGRES_DB: app
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U app -d app"]
      interval: 5s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

volumes:
  pgdata:

Apply migrations on purpose, for example with docker compose run --rm api alembic upgrade head, rather than hiding upgrade inside the container command and hoping it is safe to run on every start.

Create the engine and Redis client in the lifespan

SQLAlchemy 2.0 has async_sessionmaker. The older sessionmaker(..., class_=AsyncSession) pattern still works, but the async factory is the one to use in new code. Create one engine and one Redis client when the app starts, attach them to app.state, and close them on shutdown. redis.from_url already uses a connection pool. Set pool_size, max_overflow, and pool_pre_ping on the engine so stale connections are detected.

from contextlib import asynccontextmanager

import redis.asyncio as redis
from fastapi import FastAPI, Request
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from src.core.config import get_settings


@asynccontextmanager
async def lifespan(app: FastAPI):
    settings = get_settings()
    engine = create_async_engine(
        settings.database_url,
        pool_size=5,
        max_overflow=10,
        pool_pre_ping=True,
    )
    app.state.session_factory = async_sessionmaker(engine, expire_on_commit=False)
    app.state.redis = redis.from_url(settings.redis_url, decode_responses=True)
    await app.state.redis.ping()
    yield
    await app.state.redis.aclose()
    await engine.dispose()


async def get_db(request: Request):
    async with request.app.state.session_factory() as session:
        yield session


async def get_redis(request: Request) -> redis.Redis:
    return request.app.state.redis

expire_on_commit=False matters in async code. After a commit, you often still need attributes such as item.title to build a response. The default expired-on-commit behaviour would lazy-load those attributes and can raise MissingGreenlet in an async session. The ping() call is there so the process fails at startup if Redis is unreachable, rather than on the first cached request.

Pool settings are per process. If you run four Uvicorn workers, you have four pools. A pool_size of 5 and max_overflow of 10 can mean up to 60 connections against Postgres, whose default max_connections is often 100. Size the pool with the worker count in mind. Leave echo=True off except when you are debugging SQL.

Map a table and migrate with Alembic

Use SQLAlchemy 2.0 mapped annotations, and let Alembic own schema changes. Base.metadata.create_all() is fine for a throwaway experiment. It is not how you move a production database forward.

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String


class Base(DeclarativeBase):
    pass


class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    description: Mapped[str | None] = mapped_column(String(500), default=None)

For async URLs, Alembic should run migrations on an async engine and call the synchronous Alembic API through connection.run_sync. A trimmed env.py looks like this:

import asyncio

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

from src.core.config import get_settings
from src.models import Base

config = context.config
config.set_main_option("sqlalchemy.url", get_settings().database_url)
target_metadata = Base.metadata


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())

NullPool is intentional in Alembic: a migration should open a connection, do its work, and close, not hold an application pool. Then generate and apply a revision: alembic revision --autogenerate -m "create items" and alembic upgrade head. Review the generated script before you apply it. Autogenerate is a draft, not a guarantee.

Cache reads, invalidate on writes

The usual pattern is cache-aside. On a GET, look in Redis first. If the key is missing, load from PostgreSQL and store the JSON representation with a TTL. On a successful write, delete the key after you commit, so the next read refills from the database.

import redis.asyncio as redis
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from src.core.config import get_settings
from src.core.deps import get_db, get_redis
from src.models import Item
from src.schemas.item import ItemResponse, ItemUpdate

router = APIRouter(prefix="/items", tags=["Items"])
CACHE_PREFIX = "item:"


@router.get("/{item_id}", response_model=ItemResponse)
async def get_item(
    item_id: int,
    db: AsyncSession = Depends(get_db),
    cache: redis.Redis = Depends(get_redis),
):
    cache_key = f"{CACHE_PREFIX}{item_id}"
    cached = await cache.get(cache_key)
    if cached is not None:
        return ItemResponse.model_validate_json(cached)

    item = await db.get(Item, item_id)
    if item is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")

    payload = ItemResponse.model_validate(item)
    settings = get_settings()
    await cache.set(cache_key, payload.model_dump_json(), ex=settings.cache_ttl_seconds)
    return payload


@router.patch("/{item_id}", response_model=ItemResponse)
async def update_item(
    item_id: int,
    body: ItemUpdate,
    db: AsyncSession = Depends(get_db),
    cache: redis.Redis = Depends(get_redis),
):
    item = await db.get(Item, item_id)
    if item is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")

    for key, value in body.model_dump(exclude_unset=True).items():
        setattr(item, key, value)
    await db.commit()
    await db.refresh(item)
    await cache.delete(f"{CACHE_PREFIX}{item_id}")
    return item

Invalidate after commit, not before. If you delete the key first, another request can refill the cache from the old row while your transaction is still open. A short TTL is a backstop for that race and for any missed delete, not a substitute for invalidation. Cache a single resource first. A list endpoint is harder: one write may need to drop many keys, or you need a different key scheme.

Keep the write payload and the public response as separate Pydantic models. ItemUpdate uses optional fields so a PATCH can change only the title, and ItemResponse needs model_config = ConfigDict(from_attributes=True) so it can be built from the SQLAlchemy instance.

from pydantic import BaseModel, ConfigDict, Field


class ItemUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=100)
    description: str | None = None


class ItemResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    description: str | None = None

Do not cache user-specific data under a shared key, and do not store ORM objects in Redis. Store a JSON payload that matches that response model. If you later add a /health route, have it ping Redis and run a simple SELECT 1 against PostgreSQL. A static {"status": "ok"} will not tell you when the cache or the database is down.

What not to do

• Do not create an engine or a Redis client at import time and never close them. Use the lifespan, then aclose() and dispose().
• Do not share one AsyncSession across requests. The session is a unit of work, not a global.
• Do not use postgresql:// without +asyncpg for this engine. The sync driver will not run inside async routes the way you expect.
• Do not treat Redis as the source of truth, and do not skip invalidation because a TTL feels long enough.
• Do not copy local Compose passwords into production, or turn on SQL echo and leave it there.

Summary

A practical FastAPI, PostgreSQL, and Redis stack is mostly about ownership and lifecycle. PostgreSQL, migrated with Alembic, holds the data. Redis holds short-lived read copies that you invalidate after a write. The engine and the Redis pool belong in the lifespan, sized for the number of workers you actually run, and Docker Compose keeps the two supporting services easy to start locally.

If you want help wiring FastAPI to PostgreSQL and Redis in a production API, get in touch.