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.