Database Sessions

Published by Programming Line in FastAPI. Last updated .

A database session is the short-lived working context your FastAPI endpoint uses to talk to the database. It tracks objects you load or change, owns a transaction boundary, borrows a connection from the engine when needed, and must be closed so that connection can return to the pool. The practical outcome is simple: each request receives its own session, the endpoint commits only after its intended change succeeds, failed changes are rolled back, and the session is always closed.

In this Databases and Transactions section, sessions are the bridge between HTTP request handling and durable state. FastAPI does not create SQLAlchemy sessions for you. Instead, you define a dependency that yields a session. FastAPI enters the dependency before calling the route function, passes the yielded object into the endpoint, and runs the dependency cleanup after the response path finishes. That yield-based dependency is the usual place to guarantee close().

What The Session Actually Does

With SQLAlchemy, an Engine represents database connectivity and pooling. A Session is not the pool and usually is not a single permanent connection. It is a unit-of-work object. It keeps an identity map, so repeated loads of the same row within one session return the same Python object. It tracks pending, dirty, and deleted ORM objects. It flushes SQL statements before some queries and before commit. It opens a transaction lazily when database work begins, then either commits the transaction or rolls it back.

This matters in FastAPI because requests are concurrent. A module-level engine is normal because it is designed to manage shared connectivity. A module-level session is a bug because its identity map, transaction state, and pending writes would be shared across requests. The correct shape is engine and session factory at application scope, session instance at request scope.

API Anatomy

The moving parts are small but important. create_engine() configures the database URL, connection options, and pooling. sessionmaker() creates a factory with consistent defaults. autoflush=False avoids surprise flushes before every query, although explicit commit() still flushes pending changes. autocommit=False keeps transaction ownership explicit. The dependency creates a session, yields it, and closes it in finally. Endpoints declare db: Session = Depends(get_db) to receive the request-local session.

Example 1: Request-Scoped Session Dependency

from collections.abc import Generator

from fastapi import Depends, FastAPI
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)


def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


app = FastAPI()


@app.get("/health/db")
def db_health(db: Session = Depends(get_db)):
    return {"session_open": db.is_active}

This example establishes the core lifecycle. A request to /health/db causes FastAPI to call get_db(), receive the yielded session, pass it to the endpoint, then run the finally block. On a clean request the response body is deterministically shaped as {"session_open": true}. The value is not a full database health check, but the example proves dependency injection and request-local construction.

Example 2: Commit, Rollback, And Refresh

from collections.abc import Generator

from fastapi import Depends, FastAPI, HTTPException, status
from pydantic import BaseModel
from sqlalchemy import String, UniqueConstraint, create_engine
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmaker

DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)


class Base(DeclarativeBase):
    pass


class Widget(Base):
    __tablename__ = "widgets"
    __table_args__ = (UniqueConstraint("name", name="uq_widget_name"),)

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(80), nullable=False)


class WidgetCreate(BaseModel):
    name: str


class WidgetRead(BaseModel):
    id: int
    name: str


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


def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/widgets", response_model=WidgetRead, status_code=status.HTTP_201_CREATED)
def create_widget(payload: WidgetCreate, db: Session = Depends(get_db)):
    widget = Widget(name=payload.name)
    db.add(widget)
    try:
        db.commit()
    except IntegrityError as exc:
        db.rollback()
        raise HTTPException(status_code=409, detail="widget name already exists") from exc
    db.refresh(widget)
    return widget

Here the session tracks a new Widget. db.add() marks it pending; db.commit() flushes the insert and commits the transaction; db.refresh() reloads generated columns such as id. On a fresh database, posting {"name": "alpha"} returns status 201 and a body like {"id": 1, "name": "alpha"}. Posting the same name again violates the unique constraint, rolls back the failed transaction, and returns status 409.

Example 3: Testing With A Session Override

from collections.abc import Generator

from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

from app.main import Base, app, get_db

test_engine = create_engine(
    "sqlite://",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)
TestingSessionLocal = sessionmaker(bind=test_engine, autoflush=False, autocommit=False)
Base.metadata.create_all(bind=test_engine)


def override_get_db() -> Generator[Session, None, None]:
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()


app.dependency_overrides[get_db] = override_get_db


def test_create_widget():
    client = TestClient(app)
    response = client.post("/widgets", json={"name": "alpha"})
    assert response.status_code == 201
    assert response.json()["name"] == "alpha"

The test replaces the production dependency with an in-memory database session. StaticPool keeps the in-memory SQLite database available across connections inside the test process. The deterministic assertion is that the endpoint returns 201 and echoes the stored name. In a larger suite, clear tables between tests or create a fresh database fixture so identity and uniqueness state do not leak between cases.

Design Choices And Trade-Offs

The first major choice is where to commit. A common beginner pattern commits inside a repository helper. That is easy for single-row demos but becomes awkward when one endpoint must create several related records atomically. Prefer letting the request use case own the transaction boundary: helper functions can call add() and perform queries, while the endpoint or service layer decides when the whole unit of work commits.

The second choice is synchronous versus asynchronous database access. FastAPI supports both sync and async endpoints. A synchronous SQLAlchemy session inside a normal def endpoint is often adequate and clear. For an async database driver, use SQLAlchemy async engine and AsyncSession, and await database calls consistently. Do not mix an async endpoint with blocking database work casually; under load it can reduce event loop responsiveness.

The third choice is session lifetime. Request-scoped sessions are usually right for APIs because they match the HTTP unit of work. Longer sessions can keep stale identity-map data and hold transactions too long. Very short sessions inside each helper function make multi-step transactions harder. A background task should create its own session rather than reuse the request session after the response has been sent.

Failure Modes And Troubleshooting

Symptom: the API becomes slow and database pool errors mention checked-out connections. Likely cause: sessions are not being closed on every path. Diagnose: inspect the dependency for a finally block, check pool metrics, and reproduce with an endpoint that raises an exception. Correct: use a yield dependency and call close() in finally.

Symptom: after an insert fails, later queries using the same session raise an error about a rolled-back transaction. Likely cause: an IntegrityError occurred and the code did not call rollback(). Diagnose: look for exception handlers around commit(). Correct: catch database constraint errors where they can be translated into API responses, call rollback(), then raise an HTTP error.

Symptom: one request sees another request’s uncommitted or unexpected object state. Likely cause: a global Session instance is being reused. Diagnose: search for SessionLocal() executed at import time and shared variables named db or session. Correct: keep only the engine and factory global; instantiate the session inside the dependency.

Security, Performance, And Reliability

Sessions affect security because they are often where authorization mistakes become data leaks. Do not return ORM objects with relationships loaded accidentally unless the response model intentionally exposes them. Apply tenant filters and ownership checks before returning rows. Avoid placing credentials in the source tree; the engine should read its URL from configuration or secrets.

Performance depends on keeping transactions short. Do validation before opening expensive database work when possible. Avoid holding a transaction while calling slow external services. Use pagination for list endpoints because a session identity map containing thousands of ORM objects consumes memory. Reliability improves when database exceptions are translated deliberately: conflicts become 409, missing rows become 404, invalid input remains 422, and unexpected failures are logged without exposing internals to clients.

Hands-On Lab

Prerequisites: a FastAPI project with SQLAlchemy installed, a writable local development database, and the ability to run tests. Use a disposable SQLite file or a dedicated development database, not production data.

  1. Create an engine and SessionLocal factory in your database module.
  2. Add a get_db() dependency that creates a session, yields it, and closes it in finally.
  3. Add a small table with a unique column, such as the Widget model above.
  4. Create a POST endpoint that adds a row, commits, refreshes, and returns a response model.
  5. Catch IntegrityError, roll back, and return 409 for duplicate data.
  6. Add a test dependency override using an isolated test database.

Verification: start the app and post one unique widget. Confirm status 201 and a generated id. Post the same name again and confirm status 409. Run the test suite and confirm the endpoint works against the override database. Then temporarily raise an exception inside the endpoint and confirm the session cleanup still runs.

Cleanup: remove the disposable SQLite file, drop the development table if you used a shared database, and clear app.dependency_overrides after tests that mutate it.

Assessment Exercises

  • Design an endpoint that creates an order and several order lines. Which function should call commit(), and why?
  • A duplicate insert raises IntegrityError. Explain what happens to the session before and after rollback().
  • Find the bug in a module that defines db = SessionLocal() globally and uses it in every route. Describe the production symptom you would expect.
  • Change the test override so every test starts with empty tables. What fixture scope would you choose?
  • For a list endpoint returning thousands of rows, name two changes that reduce memory pressure in the session.

Summary

FastAPI database sessions are best understood as request-scoped units of work. The engine and session factory can live for the application lifetime, but each request should receive a fresh session. The endpoint or service layer should own the transaction boundary, call commit() only when the intended change is complete, call rollback() after failed writes, and rely on a yield dependency to close the session reliably. That discipline keeps persistence behavior predictable under concurrency, errors, and tests.