How to Add Postgres to a FastAPI App

FastAPI handles routing, validation and dependency injection, and leaves the database to you. Adding Postgres means choosing a driver or an ORM, opening a connection pool when the app starts, passing a session to each route, managing schema changes and having a Postgres server available in development, in tests and in production. This guide goes through Encore, SQLAlchemy, SQLModel and asyncpg with working code and then looks at what changes when the app is deployed.

PostgreSQL was the database the most respondents had done extensive work in, at 55.6%, in the 2025 Stack Overflow Developer Survey, which also ranked it the most desired and most admired database. In the 2024 Python Developers Survey by the PSF and JetBrains, 38% of respondents used FastAPI, up from 21% in 2021, and 35% used Django.

What you need to decide:

• Whether you write SQL directly or work through an ORM.
• Where the connection pool is created and how routes get a connection.
• Which tool owns migrations.
• Where Postgres comes from locally, in CI and in production.

Postgres options for FastAPI at a glance

Encore: add Postgres to your existing FastAPI app

Encore combines the database declaration, the local Postgres server and the production database into one line in the FastAPI app you already have. It is an infrastructure SDK rather than a new framework, so routes, dependencies and Pydantic models stay as they are, and it works the same way with Django or Flask.

The database is declared at module level, and routes query it directly:

from contextlib import asynccontextmanager

import encore
from encore.storage import postgres
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

# Encore finds this declaration, provisions the database and applies
# the SQL files in ./migrations before the service starts.
db = postgres.Database("users", migrations="./migrations")


@asynccontextmanager
async def lifespan(app: FastAPI):
    await encore.start()
    yield


app = FastAPI(lifespan=lifespan)


class NewUser(BaseModel):
    email: str


@app.post("/users", status_code=201)
async def create_user(req: NewUser) -> dict:
    row = await db.query(
        "INSERT INTO users (email) VALUES ($1) RETURNING id", req.email
    ).one()
    return {"id": row["id"]}


@app.get("/users/{user_id}")
async def get_user(user_id: int) -> dict:
    row = await db.query(
        "SELECT id, email FROM users WHERE id = $1", user_id
    ).first()
    if row is None:
        raise HTTPException(status_code=404, detail="user not found")
    return {"id": row["id"], "email": row["email"]}

Migrations are plain SQL files named by number, such as migrations/1_create_users.up.sql:

CREATE TABLE users (
    id    SERIAL PRIMARY KEY,
    email TEXT NOT NULL UNIQUE
);

CREATE TABLE audit_log (
    user_id INTEGER NOT NULL REFERENCES users(id),
    action  TEXT NOT NULL
);

Transactions use db.begin() as an async context manager, which commits when the block exits and rolls back if it raises:

@app.post("/users/{user_id}/email")
async def change_email(user_id: int, req: NewUser) -> dict:
    async with db.begin() as tx:
        await tx.exec("UPDATE users SET email = $1 WHERE id = $2", req.email, user_id)
        await tx.exec(
            "INSERT INTO audit_log (user_id, action) VALUES ($1, 'email')", user_id
        )
    return {"ok": True}

encore run starts Postgres locally, applies the migrations and starts the app, so a new developer clones the repo and runs one command. On deploy, Encore provisions the infrastructure in your AWS or GCP account, and an existing database can be imported instead of created.

A coding agent that adds a table writes the migration and the route in one change and runs both against the local database before anyone reviews it. The agent works through the code and the local tooling, while cloud access sits with Encore's IAM role, not the agent.

Without Encore

The same app without Encore needs the pieces covered in the sections below: a driver or ORM, a pool opened in the lifespan, a migration tool, a Postgres server for each developer and CI, and an RDS instance with its security group and password secret in production. That is a reasonable setup that many teams run, and it leaves more for you to own.

Good to know

Queries go through the SDK's query and exec methods with $1 style parameters rather than through an ORM session. If you want the Django ORM or another library that brings its own driver, db.connection_string hands it the connection details, which is how the Encore Django example wires up the ORM.

Try Encore →

SQLAlchemy 2.0 with asyncpg

SQLAlchemy is the most common choice, and its async extension pairs well with FastAPI. The engine holds the pool, so it is created once and disposed of on shutdown:

import os
from contextlib import asynccontextmanager
from typing import Annotated, AsyncIterator

from fastapi import Depends, FastAPI
from sqlalchemy import String, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

engine = create_async_engine(os.environ["DATABASE_URL"], pool_size=5)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String, unique=True)


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await engine.dispose()


app = FastAPI(lifespan=lifespan)


async def get_session() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        yield session


Session = Annotated[AsyncSession, Depends(get_session)]


@app.get("/users/{user_id}")
async def get_user(user_id: int, session: Session) -> dict:
    user = await session.scalar(select(User).where(User.id == user_id))
    return {"id": user.id, "email": user.email} if user else {}

DATABASE_URL looks like postgresql+asyncpg://user:pass@localhost:5432/app. Schema changes go through Alembic, which can autogenerate a migration from the difference between your models and the database.

Good to know

Each worker process gets its own pool, so the total number of connections is workers times pool_size plus overflow. Postgres allows typically 100 connections by default, which a few replicas with several workers each can reach quickly, and that is usually when teams add PgBouncer or RDS Proxy.

SQLModel

SQLModel is built on SQLAlchemy and Pydantic by the author of FastAPI, so one class can be both the table and the request or response model. Engines, sessions and Alembic work the same way as above.

from sqlmodel import Field, SQLModel


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    email: str = Field(unique=True)

Good to know

Sharing one class between the table and the API is convenient for simple CRUD and less so once the public shape of a resource differs from its storage, at which point most projects split the models again.

asyncpg directly

asyncpg is the fastest way to talk to Postgres from async Python and keeps you in plain SQL. The pool is opened in the lifespan and stored on the app state:

import os
from contextlib import asynccontextmanager

import asyncpg
from fastapi import FastAPI, Request


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.pool = await asyncpg.create_pool(os.environ["DATABASE_URL"])
    yield
    await app.state.pool.close()


app = FastAPI(lifespan=lifespan)


@app.get("/users/{user_id}")
async def get_user(user_id: int, request: Request) -> dict:
    row = await request.app.state.pool.fetchrow(
        "SELECT id, email FROM users WHERE id = $1", user_id
    )
    return dict(row) if row else {}

Good to know

asyncpg has no migration tool, so you add one yourself, and the local and production databases are still yours to run.

Running Postgres in production

Whichever library you choose, production needs a managed database, credentials stored outside the code and a network path from the service to the database. On AWS that is usually RDS with the password in Secrets Manager, a security group that only admits the service and a task role allowed to read the secret. With Encore, you connect your own cloud account and Encore provisions the infrastructure in it on deploy, so the database, its credentials and the network path are not written by hand.

How to choose

• Encore if you want the database running locally and in your AWS or GCP account from the declaration in your FastAPI code.
• SQLAlchemy with Alembic for an ORM, relationships and autogenerated migrations.
• SQLModel for CRUD-heavy apps where table and API models line up.
• asyncpg for hand-written SQL with the least overhead.

Python support in Encore is in beta, and you can sign up for access to try it with your FastAPI app.

Related guides

• Deploy FastAPI to AWS
• FastAPI vs Django vs Flask
• Best Python backend frameworks
• Python microservices
• FastAPI vs Django
• Litestar vs FastAPI
• How to deploy a Django app to AWS with an AI coding agent