Database Connections and Sessions

You've built a beautiful FastAPI app with endpoints, Pydantic models, and dependency injection — but the moment you add a database, everything slows down or breaks under load. Connections get exhausted, sessions leak, and your API returns 500 errors at the worst possible moment. Let's fix that by mastering database connections and sessions the FastAPI way.

The problem this lesson solves

Every time your API handles a request, it needs to talk to your database. Doing that naively — opening a new connection per request, or sharing a single global connection — leads to two classic failures: connection exhaustion (the database runs out of connections and refuses new ones) and race conditions (interleaved queries corrupt data). You need a pattern that gives each request its own safe, short-lived database session while reusing expensive connections underneath.

This lesson gives you that pattern using SQLAlchemy with FastAPI's dependency injection system. You'll learn why connections and sessions are different, how to manage them cleanly, and what to do when things go wrong.

Core concept / mental model

Think of a database connection as a physical phone line to your database server. Establishing it costs time and memory. A session is a conversation over that line — you open it, run your queries, commit or roll back, and hang up. You never want to hold one phone line open forever, nor do you want a thousand people fighting over a single line.

Here's the mental model in three layers:

1. Engine — the switchboard. It manages a pool of reusable connections.
2. Session — one conversation. Created per request, used, then closed.
3. Dependency — FastAPI's way to hand each request its own session and guarantee cleanup.

Pro tip: A session is not a connection. A session may use multiple connections during its lifetime, or none at all if all data is cached. Always scope your session to the request.

How it works step by step

Follow this sequence to set up database connections and sessions in FastAPI:

1. Install dependencies: sqlalchemy and optionally psycopg2-binary for PostgreSQL, or keep the built-in SQLite for local development.
2. Create a database engine — SQLAlchemy's create_engine() manages your connection pool.
3. Define a SessionLocal factory — bound to your engine, this creates new sessions on demand.
4. Create a dependency function — get_db() yields a session, FastAPI closes it after the request.
5. Use the dependency in your routes — inject Session into your endpoint, run queries, commit, and let FastAPI handle cleanup.

Cause and effect: When a request hits your endpoint, FastAPI calls get_db(), you get a fresh session, you run your business logic, and the finally block ensures the session is closed — even if an exception occurs. Without the finally, connections leak and eventually exhaust your database.

Hands-on walkthrough

Let's build a minimal but complete example. Start with a database.py module:

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# PostgreSQL example — swap for SQLite if you prefer
DATABASE_URL = "postgresql://user:password@localhost/mydb"

engine = create_engine(
    DATABASE_URL,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,
)

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

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Now define a simple model:

# models.py
from sqlalchemy import Column, Integer, String
from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

Wire it into your FastAPI app:

# main.py
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from database import get_db, Base, engine
from models import User
from pydantic import BaseModel

app = FastAPI()
Base.metadata.create_all(bind=engine)

class UserCreate(BaseModel):
    name: str
    email: str

@app.post("/users/", response_model=UserCreate)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Run it with uvicorn main:app --reload — you now have a working, session-managed API.

Expected output

When you POST a new user, you get the created object back with all fields populated. A GET returns the stored user or a 404 if missing. Each request opens its own session, commits on success, and closes automatically.

Compare options / when to choose what

For most CRUD apps, the first option is your default. If you're handling long-running background tasks, create separate sessions per task. If you're building a heavily async service, consider async SQLAlchemy with async_sessionmaker.

Troubleshooting & edge cases

Error: sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) FATAL: remaining connection slots are reserved

Your connection pool is exhausted. Either you're not closing sessions, or your pool size is too small. Check that your finally block closes sessions — or use FastAPI's dependency with yield. Increase pool_size and max_overflow if needed.

Error: sqlalchemy.exc.InterfaceError: connection already closed

A connection died, often due to a firewall timeout. Add pool_pre_ping=True to your engine — SQLAlchemy will check the connection before using it and reopen if needed.

Sessions not committing

You called db.add() but forgot db.commit(). SQLAlchemy autoflush is off in our setup, so changes aren't sent without explicit commit. Always commit at the end of a successful transaction.

Session is used outside request scope

You stored db in a global or background task. Sessions are tied to a specific moment — a connection is bound at first use. Don't pass sessions between threads or coroutines.

Pro tip: Set autoflush=False to avoid surprise SELECT statements before your commit(). It also makes your code more predictable.

What you learned & what's next

You now understand:

• The difference between connections (reusable, pooled) and sessions (per-request conversations)
• How to create an engine with a connection pool and pool_pre_ping
• How to build a get_db() dependency that guarantees session cleanup
• How to inject sessions into routes via Depends
• How to compare approaches and pick the right one for your use case

You're ready to take the next step in your FastAPI journey: dependency injection and middleware. You'll learn how to share objects like database sessions across your entire app, handle authentication, and add request logging — all with clean, testable code. This database foundation will be the bedrock of every future API you build.

Happy building!