Data and State

FastAPI is not tied to any particular database or ORM. This chapter connects a database with SQLModel, the library the official docs use in their examples, manages sessions through dependencies, reads configuration with pydantic-settings, and runs work after the response with background tasks.

SQLModel and SQLAlchemy

SQLModel, written by FastAPI's author, layers Pydantic on top of SQLAlchemy so a single class can define both a table and an API model. Using SQLAlchemy directly is also a common choice, and the pattern of injecting the session as a dependency is the same.

pip install sqlmodel

Models and the engine

A class with table=True becomes a database table; other classes are plain data models for input and output. Putting shared fields in a base class and inheriting from it avoids repetition.

# app/db.py
from typing import Annotated

from fastapi import Depends
from sqlmodel import Field, Session, SQLModel, create_engine


class HeroBase(SQLModel):
    name: str = Field(index=True)
    age: int | None = None


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    secret_name: str


class HeroCreate(HeroBase):
    secret_name: str


class HeroPublic(HeroBase):
    id: int


engine = create_engine(
    "sqlite:///database.db",
    connect_args={"check_same_thread": False},  # only needed for SQLite
)


def get_session():
    with Session(engine) as session:
        yield session


SessionDep = Annotated[Session, Depends(get_session)]

get_session is a dependency that uses yield. Each request gets its own session, passed into the path operation, and the with block closes it once the response is done. You can create tables at startup in lifespan with SQLModel.metadata.create_all(engine), but in production a migration tool such as Alembic is the better choice.

CRUD endpoints

With the session dependency in place, reads, inserts and lookups are short. Because the response model is HeroPublic, secret_name never appears in responses.

from typing import Annotated

from fastapi import FastAPI, HTTPException, Query
from sqlmodel import select

from app.db import Hero, HeroCreate, HeroPublic, SessionDep

app = FastAPI()


@app.post("/heroes", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero


@app.get("/heroes", response_model=list[HeroPublic])
def read_heroes(
    session: SessionDep,
    offset: int = 0,
    limit: Annotated[int, Query(le=100)] = 100,
):
    return session.exec(select(Hero).offset(offset).limit(limit)).all()


@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

This example uses a synchronous session, so the path operations are plain def. With an async driver such as asyncpg and SQLAlchemy's AsyncSession, you would use async def and await instead.

Settings with pydantic-settings

Pass values like the database URL and secret key through environment variables, and build a typed settings object with BaseSettings from pydantic-settings. Environment variable names map to field names case-insensitively.

# app/config.py
from functools import lru_cache

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    app_name: str = "Hello FastAPI"
    database_url: str = "sqlite:///database.db"
    secret_key: str

    model_config = SettingsConfigDict(env_file=".env")


@lru_cache
def get_settings() -> Settings:
    return Settings()

Install it with pip install pydantic-settings. A field without a default, like secret_key, raises a validation error at startup if the variable is missing, so misconfiguration surfaces early. @lru_cache means the .env file is read only once; in routes, receive it as Annotated[Settings, Depends(get_settings)]. Injecting it as a dependency also makes it easy to swap in different settings during tests.

Background tasks

Work the client should not wait for, such as sending email or writing logs, can run after the response is sent using BackgroundTasks:

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message + "\n")


@app.post("/send-notification/{email}")
def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"notification sent to {email}")
    return {"message": "Notification queued"}

Background tasks run inside the same process, so they suit lightweight jobs. For long-running work, jobs that need retries, or work that must survive a server restart, use a dedicated task queue such as Celery or ARQ.

Summary

• Open and close a database session per request with a yield dependency.
• Use SQLModel to define table models separately from input and output models.
• Read configuration with BaseSettings from pydantic-settings and a .env file, injected as a dependency.
• Run small post-response jobs with BackgroundTasks and heavy ones in a task queue.