Connecting PostgreSQL to FastAPI means giving your API a real database behind its endpoints, so data survives restarts and many users can read and write it at once. This tutorial builds a small books API with FastAPI, SQLAlchemy 2.0, and psycopg 3. Every command below was run on this machine against PostgreSQL 16.2, with the real output shown.
Key takeaways
- SQLAlchemy 2.0 with the postgresql+psycopg dialect is the standard way to connect FastAPI to PostgreSQL, as shown in the official FastAPI SQL databases tutorial.
- The engine is created once at startup. Each request gets its own session through a FastAPI dependency that closes it when the request ends.
- pool_pre_ping drops dead connections before use, which stops stale-connection errors after idle time.
- I ran every step against PostgreSQL 16.2 with FastAPI 0.162.2, SQLAlchemy 2.1.8, and psycopg 3.3.8 on Python 3.12.
- Keep the connection string in an environment variable, never in code you commit.
What are we building?
A books API with four endpoints. GET /health proves the database answers. POST /books adds a book. GET /books lists every book. GET /books/{book_id} fetches one book or returns a 404. The data lives in a PostgreSQL table called books, and SQLAlchemy maps a Python class to that table. By the end you will have a pattern you can reuse for almost any FastAPI service: one engine, one session dependency, and plain CRUD endpoints on top.
My opinion, up front: do not write raw SQL strings in your route handlers. SQLAlchemy is not there to hide SQL from you. It is there so connection handling, parameter binding, and transactions happen in one place, and your routes stay about HTTP. The moment you need a join or a filter, you will be glad the query is built from objects instead of string concatenation.
What do you need before you start?
- Python 3.10 or newer. I used Python 3.12.
- PostgreSQL 16 running locally, or a hosted instance you can reach. The steps below use a local server on port 5432.
- A terminal and a text editor. No Docker is needed for this tutorial, though the Docker one-liner in Step 1 works if you prefer it.
Step 1: How do you set up PostgreSQL?
First, check the server answers. If you installed PostgreSQL from your OS packages, start it, then run this check with the psycopg driver or psql:
psql -U postgres -c "SELECT version();"
On my machine this printed:
PostgreSQL 16.2 on x86_64-pc-linux-gnu, compiled by gcc (GCC) 10.2.1 20210130 (Red Hat 10.2.1-11), 64-bit
Next, create a database for the tutorial. SQLAlchemy can create tables, but it will not create the database itself, so do that first:
createdb -U postgres tutorial_db
If you prefer Docker, one command gives you the same thing: docker run --name tutorial-pg -e POSTGRES_PASSWORD=secret -p 5432:5432 -d postgres:16, then createdb inside it. Either way, the rest of the tutorial is identical. The database name is tutorial_db and the server listens on localhost:5432.
Step 2: How do you install the Python packages?
Create a project folder and a virtual environment, then install four packages:
mkdir fastapi-postgres && cd fastapi-postgres
python -m venv .venv
source .venv/bin/activate
pip install fastapi "uvicorn[standard]" sqlalchemy "psycopg[binary]"
Each package has one job. FastAPI serves the HTTP API. Uvicorn is the server that runs it. SQLAlchemy talks to the database. Psycopg 3 is the driver that carries SQLAlchemy's queries to PostgreSQL over the wire. Note the package name: psycopg 3 is installed as psycopg, and the [binary] extra ships a prebuilt wheel so you do not need a C compiler. The older driver, psycopg2, still works, but psycopg 3 is the current line and the one SQLAlchemy 2.0 pairs with in its docs.
Pin the result in a requirements.txt so the next person gets the same versions I tested:
fastapi==0.162.2
uvicorn==0.54.0
SQLAlchemy==2.1.8
psycopg[binary]==3.3.8
Step 3: How do you connect SQLAlchemy to PostgreSQL?
Create a file called app.py. The first block is the whole connection, and it is the most important part of this tutorial:
import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
DATABASE_URL = os.environ.get(
"DATABASE_URL",
"postgresql+psycopg://postgres@localhost:5432/tutorial_db",
)
engine = create_engine(
DATABASE_URL, pool_size=5, max_overflow=10, pool_pre_ping=True
)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
Base = declarative_base()
Read the connection string left to right. postgresql+psycopg picks the dialect and the driver. postgres is the database user. localhost:5432 is the host and port. tutorial_db is the database from Step 1. If your server needs a password, the string becomes postgresql+psycopg://postgres:YOUR_PASSWORD@localhost:5432/tutorial_db. Put the real value in the DATABASE_URL environment variable and keep the default as a local-dev fallback only.
Three engine settings matter. pool_size=5 keeps five connections open for reuse. max_overflow=10 lets ten extra connections open under a burst, then closes them. pool_pre_ping=True sends a tiny check before handing a connection to your code, so a connection the database closed while idle is replaced instead of failing your request. The engine opens connections lazily: nothing touches the database until the first query runs, which is why a typo in the URL only shows up when you hit an endpoint.
Step 4: How do you model the table?
Add the model and the Pydantic schemas to app.py. The model is the table. The schemas are the shapes the API accepts and returns:
from sqlalchemy import Column, Integer, String
from pydantic import BaseModel
class Book(Base):
__tablename__ = "books"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, nullable=False)
author = Column(String, nullable=False)
year = Column(Integer, nullable=True)
class BookCreate(BaseModel):
title: str
author: str
year: int | None = None
class BookOut(BookCreate):
id: int
Two details deserve attention. First, BookCreate and BookOut are separate on purpose: the client sends no id, and the server adds it. One schema for both directions invites clients to set ids they do not own. Second, the table is created at startup, not by hand. Add a lifespan handler that runs Base.metadata.create_all(bind=engine) once when the app starts. For a tutorial this is fine. For a real service, use Alembic migrations instead, because create_all never alters an existing table.
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
Base.metadata.create_all(bind=engine)
yield
app = FastAPI(title="Books API", lifespan=lifespan)
Step 5: How do you wire the session into FastAPI?
This is the dependency that every endpoint will use:
from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
FastAPI calls get_db before the endpoint runs, hands the session in, and resumes the generator after the response is sent, which runs db.close(). That finally is the whole point. A session that is never closed holds a pooled connection forever, and a busy API will run out of connections one request at a time. Never create a session at module level and share it across requests. One request, one session, closed in finally.
Now the endpoints. Each one takes db: Session = Depends(get_db) and does one small thing:
from sqlalchemy import text
@app.get("/health")
def health(db: Session = Depends(get_db)):
db.execute(text("SELECT 1"))
return {"status": "ok", "database": "connected"}
@app.post("/books", response_model=BookOut, status_code=201)
def create_book(book: BookCreate, db: Session = Depends(get_db)):
new_book = Book(**book.model_dump())
db.add(new_book)
db.commit()
db.refresh(new_book)
return new_book
@app.get("/books", response_model=list[BookOut])
def list_books(db: Session = Depends(get_db)):
return db.query(Book).order_by(Book.id).all()
@app.get("/books/{book_id}", response_model=BookOut)
def get_book(book_id: int, db: Session = Depends(get_db)):
book = db.query(Book).filter(Book.id == book_id).first()
if book is None:
raise HTTPException(status_code=404, detail="Book not found")
return book
The health endpoint is not decoration. It runs SELECT 1 through the same engine and session as every other route, so if it returns connected, the database path works. Load balancers and uptime monitors can hit it without touching your data.
Step 6: How do you test the API?
Start the server:
uvicorn app:app --host 127.0.0.1 --port 8000
The startup log should end with these two lines:
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
Then exercise every endpoint with curl. First the health check:
curl -s http://127.0.0.1:8000/health
{"status":"ok","database":"connected"}
Add two books:
curl -s -X POST http://127.0.0.1:8000/books \
-H "Content-Type: application/json" \
-d '{"title": "Designing Data-Intensive Applications", "author": "Martin Kleppmann", "year": 2017}'
{"title":"Designing Data-Intensive Applications","author":"Martin Kleppmann","year":2017,"id":1}
curl -s -X POST http://127.0.0.1:8000/books \
-H "Content-Type: application/json" \
-d '{"title": "The Pragmatic Programmer", "author": "Andrew Hunt", "year": 1999}'
{"title":"The Pragmatic Programmer","author":"Andrew Hunt","year":1999,"id":2}
List them, fetch one, and ask for a book that does not exist:
curl -s http://127.0.0.1:8000/books
[{"title":"Designing Data-Intensive Applications","author":"Martin Kleppmann","year":2017,"id":1},{"title":"The Pragmatic Programmer","author":"Andrew Hunt","year":1999,"id":2}]
curl -s http://127.0.0.1:8000/books/1
{"title":"Designing Data-Intensive Applications","author":"Martin Kleppmann","year":2017,"id":1}
curl -s http://127.0.0.1:8000/books/99
{"detail":"Book not found"}
That last call returns HTTP 404, which is the correct answer for a missing row. I also confirmed the rows are really in PostgreSQL, not just in memory, by querying the table directly:
(1, 'Designing Data-Intensive Applications', 'Martin Kleppmann', 2017)
(2, 'The Pragmatic Programmer', 'Andrew Hunt', 1999)
Open http://127.0.0.1:8000/docs in a browser and you get the interactive Swagger UI for free, with a form for each endpoint. FastAPI builds it from your schemas, so the docs stay in step with the code.
Why does the connection pool matter?
Opening a fresh PostgreSQL connection costs a TCP handshake, authentication, and a backend process on the server. Doing that on every request is slow and, under load, it exhausts the server's connection limit. The pool pays that cost once and reuses the connection. Five pooled connections will serve far more than five requests per second, because each request borrows one for milliseconds and returns it.
The failure mode is worth knowing. If a request forgets to close its session, the pool slowly drains, and the next request hangs waiting for a connection that never comes back. That is why the try/finally in get_db is not optional style. It is the fix for the most common FastAPI production incident with a database.
What breaks most often?
Connection refused. PostgreSQL is not running, or it listens on a different port. Check with psql -U postgres -c "SELECT 1;" before blaming FastAPI.
Password authentication failed. The user or password in the URL is wrong. Local installs often use trust auth with no password, while Docker and hosted databases need one. Put the full URL in DATABASE_URL and test it with psql first.
ModuleNotFoundError: No module named 'psycopg'. You installed the driver in a different environment than the one running uvicorn. Activate the venv first, then run both pip install and uvicorn from it.
Driver name wrong in the URL. SQLAlchemy 2.0 needs postgresql+psycopg for psycopg 3. Writing postgresql+psycopg2 while only psycopg 3 is installed fails at engine creation with a clear error naming the missing driver.
Table does not exist. The lifespan handler only creates tables at startup. If you added a model after the server started, restart it. If you changed a column, create_all will not alter the table: drop it or switch to Alembic.
What to try next
- Add a GET /books?author=Kleppmann filter with db.query(Book).filter(Book.author == author), and a DELETE /books/{book_id} endpoint that returns 404 when the row is missing.
- Move the table creation to Alembic migrations, so changing a column no longer means dropping the table.
- Deploy it: the same DATABASE_URL pattern works on Render with a managed PostgreSQL instance, and the health endpoint is ready for the platform's health checks.
Originally published on LogicStack. Run the code examples right in your browser with our free interactive Python and SQL playgrounds — no setup, no sign-up.
Top comments (0)