SQLAlchemy Setup — Engine, Sessions & Models
IntermediateOne engine per app (with a connection pool), one session per request (via the get_db yield dependency), ORM models declaring tables in Python — the wiring every FastAPI + Postgres service shares.
Overview
FastAPI has no ORM of its own; the ecosystem standard is SQLAlchemy 2.x. Three objects run everything: the engine (created once, owns the connection pool), sessionmaker (a factory for sessions), and the session (your unit of work — short-lived, one per request). ORM models are classes mapped to tables via Mapped[] annotations, and the get_db yield dependency from the DI chapters hands each request its own session and guarantees it closes. The cardinal rules are about lifetimes: engine per application, session per request, never share a session across requests or threads. Pool sizing is the production knob people discover during their first traffic spike — and pool_pre_ping is the flag that saves you from stale connections after the DB restarts.
Engine, Models, Session Factory
The engine is created once at import with pool settings; models map tables with typed columns; sessionmaker stamps out sessions. SQLAlchemy 2.x style throughout — Mapped and mapped_column, not the legacy Column.
# pip install sqlalchemy psycopg2-binary
# ── app/core/db.py ──────────────────────────
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker
engine = create_engine(
"postgresql://app:app@localhost:5432/placement", # from Settings in real code
pool_size=5, # steady connections held open
max_overflow=10, # extra under burst (returned when idle)
pool_pre_ping=True, # test connection before use — survives DB restarts
echo=False, # True in dev = log every SQL statement
)
SessionLocal = sessionmaker(bind=engine, autoflush=False)
class Base(DeclarativeBase):
pass
# ── app/models/student.py ───────────────────
from sqlalchemy import String, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
class Student(Base):
__tablename__ = "students"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(60))
email: Mapped[str] = mapped_column(String(120), unique=True, index=True)
cgpa: Mapped[float] = mapped_column(default=0.0)
branch: Mapped[str] = mapped_column(String(10), index=True)
offers: Mapped[list["Offer"]] = relationship(back_populates="student")
class Offer(Base):
__tablename__ = "offers"
id: Mapped[int] = mapped_column(primary_key=True)
student_id: Mapped[int] = mapped_column(ForeignKey("students.id"), index=True)
company: Mapped[str] = mapped_column(String(80))
ctc_lpa: Mapped[float]
student: Mapped[Student] = relationship(back_populates="offers")
# Dev bootstrap only — real schema changes go through Alembic:
# Base.metadata.create_all(engine)Session per Request — get_db for Real
The yield dependency owns the session lifecycle. Endpoints receive a live session, queries use the 2.x select() style, and the lifetime rules below are the difference between stable and mysteriously broken.
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.db import SessionLocal
from app.models.student import Student
app = FastAPI()
def get_db():
db = SessionLocal() # one session for THIS request
try:
yield db
db.commit()
except Exception:
db.rollback()
raise
finally:
db.close() # returns connection to the pool
@app.get("/students/{student_id}")
def get_student(student_id: int, db: Session = Depends(get_db)):
student = db.get(Student, student_id) # PK lookup
if not student:
raise HTTPException(404, "student not found")
return {"id": student.id, "name": student.name, "cgpa": student.cgpa}
@app.get("/students")
def top_students(db: Session = Depends(get_db)):
rows = db.execute(
select(Student)
.where(Student.cgpa >= 8.0, Student.branch == "CS")
.order_by(Student.cgpa.desc())
.limit(10)
).scalars().all()
return [{"id": s.id, "name": s.name} for s in rows]
# Lifetime rules (interview gold):
# engine → ONE per application (owns the pool)
# session → ONE per request (cheap; borrows a pooled connection)
# sharing a session across requests/threads → race conditions, stale data
# Sizing: pool_size + max_overflow ≥ workers × concurrent DB calls,
# and ≤ Postgres max_connections across ALL app instances.Key Points to Remember
- 1Engine created once with the pool; sessionmaker stamps out per-request sessions
- 2get_db yield dependency: commit on success, rollback on error, always close
- 3SQLAlchemy 2.x: Mapped[]/mapped_column models, select() queries, db.get() for PKs
- 4pool_pre_ping=True survives DB restarts; size pools against max_connections
Interview Questions
Sign in to ask AriaWhy one engine per app but one session per request? What breaks otherwise?
What do pool_size, max_overflow, and pool_pre_ping control?
Your API works for weeks, then every request 500s after a DB failover — likely cause and fix?
Ask Aria about SQLAlchemy Setup — Engine, Sessions & Models
Your personal AI tutor — ask anything about this concept
Revision Status
Personal Notes
Sign in to save personal notes for this topic.
Discussion
Sign in to join the discussion.