Database Sessions

Difficulty: Intermediate

One of the most important uses of FastAPI's dependency injection is managing database sessions. A database session should be created at the start of a request, used by the handler and its dependencies, and properly closed when the request finishes - even if an error occurs. FastAPI's yield dependencies make this pattern clean and reliable.

A yield dependency is a generator function that yields a value instead of returning it. FastAPI calls the function up to the yield point, passes the yielded value to the handler, waits for the handler to complete, and then executes the code after the yield. This after-yield code runs even if the handler raised an exception, making it perfect for cleanup operations like closing database sessions.

The pattern is similar to Python's context managers (with statement). The code before yield is the setup phase, the yielded value is what the handler receives, and the code after yield is the teardown phase. You can use try/finally to ensure cleanup happens regardless of exceptions.

In a typical SQLAlchemy setup, the get_db dependency creates a session, yields it to the handler, and closes it in the finally block. The handler uses the session for database operations. If an exception occurs, the finally block still closes the session. If the operation succeeds, you can commit the transaction before yielding (or let the handler commit).

For async database libraries (like asyncpg or SQLAlchemy async), the yield dependency should be an async generator (async def with yield). This allows proper async cleanup of database resources.

Yield dependencies can also be used for other resource management patterns: file handles, locks, temporary files, connections to external services, and anything else that needs cleanup after use.

Code examples

Basic Yield Dependency

from fastapi import FastAPI, Depends

app = FastAPI()

# Simulated database session
class FakeDBSession:
    def __init__(self):
        print("Opening database session")
        self.data = {1: "Alice", 2: "Bob"}

    def get(self, key):
        return self.data.get(key)

    def close(self):
        print("Closing database session")

def get_db():
    db = FakeDBSession()
    try:
        yield db  # Handler receives this
    finally:
        db.close()  # Always runs, even on errors

@app.get("/users/{user_id}")
def get_user(user_id: int, db: FakeDBSession = Depends(get_db)):
    name = db.get(user_id)
    if not name:
        raise HTTPException(status_code=404)
    return {"id": user_id, "name": name}

# Request flow:
# 1. get_db() creates FakeDBSession
# 2. Yields session to get_user handler
# 3. Handler executes
# 4. get_db() finally block closes session

The yield keyword splits the dependency into setup (before yield) and teardown (after yield). The try/finally ensures the session is closed even if the handler raises an exception. This is the standard pattern for database session management.

SQLAlchemy Session Dependency

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

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

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

app = FastAPI()

@app.get("/users")
def list_users(db: Session = Depends(get_db)):
    # Use the db session for queries
    # db.query(User).all()
    return [{"id": 1, "name": "Alice"}]

@app.post("/users")
def create_user(db: Session = Depends(get_db)):
    # db.add(user)
    # db.commit()
    # db.refresh(user)
    return {"id": 1, "name": "Alice"}

# Every endpoint that depends on get_db gets a fresh session
# The session is automatically closed after the request

This is the standard SQLAlchemy session management pattern in FastAPI. SessionLocal creates new sessions, get_db yields a session and ensures it is closed. Every endpoint that needs database access depends on get_db.

Async Database Session

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

# Async database setup
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
async_engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(
    async_engine, class_=AsyncSession, expire_on_commit=False
)

# Async yield dependency
async def get_async_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

app = FastAPI()

@app.get("/users")
async def list_users(db: AsyncSession = Depends(get_async_db)):
    # result = await db.execute(select(User))
    # users = result.scalars().all()
    return [{"id": 1, "name": "Alice"}]

Async sessions use async generators (async def with yield). The async with statement handles session lifecycle. Commit on success, rollback on exception. This works with asyncpg, aiosqlite, and other async database drivers.

Key points

Concepts covered

yield Dependencies, Session Management, Cleanup, Context Manager, Generator