Skillversitas

SQL Databases

FastAPI pairs naturally with SQL databases through SQLAlchemy, the de-facto ORM for Python. The usual architecture: define models, create an engine and session, pass the session into endpoints via a dependency, and query with the ORM or Core.

Install

pip install fastapi sqlalchemy "uvicorn[standard]"

Define Models

from sqlalchemy import create_engine, String, Float
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    price: Mapped[float] = mapped_column(Float)

engine = create_engine("sqlite:///./app.db", connect_args={"check_same_thread": False})
Base.metadata.create_all(engine)

The Session Dependency

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

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

app = FastAPI()

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

Each request gets a fresh session that is closed in the finally block - no connection leaks.

Querying with the ORM

@app.post("/items/")
def create_item(name: str, price: float, db: Session = Depends(get_db)):
    item = Item(name=name, price=price)
    db.add(item)
    db.commit()
    db.refresh(item)
    return item

@app.get("/items/{item_id}")
def read_item(item_id: int, db: Session = Depends(get_db)):
    item = db.get(Item, item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Not found")
    return item

@app.delete("/items/{item_id}")
def delete_item(item_id: int, db: Session = Depends(get_db)):
    item = db.get(Item, item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Not found")
    db.delete(item)
    db.commit()
    return None

Practical Notes

  • For SQLite with FastAPI, connect_args check_same_thread=False is needed because requests arrive on different threads.
  • For PostgreSQL use a postgresql+psycopg:// URL and run Alembic migrations instead of create_all in production.
  • Use from_attributes=True on response models to serialize ORM objects directly.

Summary

Connect FastAPI to SQL with SQLAlchemy: define declarative models, build an engine, expose a session as a Depends() dependency that opens and closes per request, and perform CRUD with db.add/get/query/commit. Use migrations for schema changes and response models to keep the output contract clean.