Difficulty: Advanced
SQLAlchemy is the most popular Python ORM (Object-Relational Mapper), and it integrates seamlessly with FastAPI. An ORM lets you interact with your database using Python classes and objects instead of raw SQL queries. SQLAlchemy 2.0 introduced a modern syntax with improved type support that works particularly well with FastAPI and Pydantic.
The first step in setting up SQLAlchemy is creating an engine. The engine is the starting point for all SQLAlchemy operations - it manages the connection pool and provides connectivity to the database. You create it with create_engine(DATABASE_URL), where the URL follows the format: dialect+driver://user:password@host:port/dbname. For PostgreSQL, this looks like postgresql://user:pass@localhost:5432/mydb. For SQLite, it is sqlite:///./app.db.
A session is the workspace for your database operations. It tracks changes to objects, batches SQL statements, and manages transactions. You create sessions using sessionmaker, which is a factory bound to your engine. Each request should get its own session (via a FastAPI dependency) to prevent cross-request interference.
SQLAlchemy 2.0 uses DeclarativeBase (replacing the older declarative_base() function) to define your ORM models. Each model class represents a database table. Columns are defined using mapped_column() with type annotations, replacing the older Column() syntax. Relationships between tables are defined using relationship().
The connection between SQLAlchemy models (database representation) and Pydantic models (API representation) is important. SQLAlchemy models define the database schema. Pydantic models define the API request/response schemas. You typically convert between them in your CRUD operations. Pydantic's model_config with from_attributes=True enables creating Pydantic models directly from SQLAlchemy objects.
For development, SQLAlchemy can create tables automatically using Base.metadata.create_all(engine). For production, use Alembic migrations to manage schema changes incrementally and safely.
from sqlalchemy import create_engine, String, Integer
from sqlalchemy.orm import (
DeclarativeBase, Mapped, mapped_column, sessionmaker,
)
# Database URL
DATABASE_URL = "sqlite:///./app.db"
# Create engine
engine = create_engine(
DATABASE_URL,
connect_args={"check_same_thread": False}, # SQLite only
echo=True, # Log SQL statements (dev only)
)
# Session factory
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Base class for models
class Base(DeclarativeBase):
pass
# ORM Model
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True, index=True)
username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
email: Mapped[str] = mapped_column(String(100), unique=True)
is_active: Mapped[bool] = mapped_column(default=True)
# Create tables
Base.metadata.create_all(bind=engine)The engine manages database connections. SessionLocal creates new sessions. Base is the parent class for all ORM models. The User model maps to a 'users' table. mapped_column replaces the older Column() syntax in SQLAlchemy 2.0.
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from pydantic import BaseModel, ConfigDict
# ... (engine, SessionLocal, Base, User from above)
app = FastAPI()
# Database session dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# Pydantic schemas (separate from ORM models)
class UserCreate(BaseModel):
username: str
email: str
class UserResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
username: str
email: str
is_active: bool
@app.post("/users", response_model=UserResponse, status_code=201)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
db_user = User(username=user.username, email=user.email)
db.add(db_user)
db.commit()
db.refresh(db_user) # Refresh to get the auto-generated ID
return db_user # SQLAlchemy model -> Pydantic via from_attributes
@app.get("/users", response_model=list[UserResponse])
def list_users(db: Session = Depends(get_db)):
return db.query(User).all()The get_db dependency provides a session per request and ensures cleanup. Pydantic schemas with from_attributes=True can be created directly from SQLAlchemy models. The handler returns SQLAlchemy objects, and FastAPI converts them using the response_model.
# database.py
import os
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker
DATABASE_URL = os.getenv(
"DATABASE_URL",
"postgresql://postgres:password@localhost:5432/myapp"
)
engine = create_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
class Base(DeclarativeBase):
pass
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# models.py
from sqlalchemy import String, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from database import Base
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
username: Mapped[str] = mapped_column(String(50), unique=True)
email: Mapped[str] = mapped_column(String(100))
posts: Mapped[list["Post"]] = relationship(back_populates="author")
class Post(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200))
body: Mapped[str] = mapped_column()
author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
author: Mapped["User"] = relationship(back_populates="posts")Separating database configuration into its own module keeps the code organized. pool_pre_ping=True checks connections before use, preventing stale connection errors. The User-Post relationship demonstrates one-to-many with back_populates.
SQLAlchemy, ORM, Engine, Session, DeclarativeBase