Database session management in FastAPI

Introduction

Database session management helps your app talk to the database safely and efficiently. It keeps track of changes and makes sure data is saved or rolled back properly.

When you want to save user data from a web form into a database.

When you need to read or update records in a database during a web request.

When you want to make sure database changes happen only if all steps succeed.

When you want to avoid leaving open connections that waste resources.

When you want to handle errors and rollback changes if something goes wrong.

Syntax

FastAPI

from sqlalchemy.orm import Session
from fastapi import Depends

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

This function creates a database session and closes it after use.

Use Depends(get_db) in your path operation to get the session.

Examples

This example shows how to get a database session in a route and query all items.

FastAPI

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

app = FastAPI()

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

@app.get("/items/")
def read_items(db: Session = Depends(get_db)):
    items = db.query(Item).all()
    return items

This example shows how to add a new item to the database using the session.

FastAPI

from fastapi import Depends
from sqlalchemy.orm import Session

@app.post("/items/")
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = Item(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

Sample Program

This FastAPI app connects to a SQLite database. It manages sessions with get_db. You can add items and list all items. The session opens for each request and closes after.

FastAPI

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base, Session

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)

Base.metadata.create_all(bind=engine)

app = FastAPI()

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

@app.post("/items/")
def create_item(name: str, db: Session = Depends(get_db)):
    db_item = Item(name=name)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return {"id": db_item.id, "name": db_item.name}

@app.get("/items/")
def read_items(db: Session = Depends(get_db)):
    items = db.query(Item).all()
    return [{"id": item.id, "name": item.name} for item in items]

Important Notes

Always close the session to avoid database connection leaks.

Use commit() to save changes and refresh() to get updated data like auto-generated IDs.

Use yield in get_db to create a clean way to open and close sessions per request.

Summary

Database session management helps your app safely talk to the database.

Use a function like get_db to open and close sessions for each request.

Use the session to add, query, and commit data inside your FastAPI routes.