SQLAlchemy — Python's Database Toolkit
AdvancedSQLAlchemy maps classes to tables (ORM), builds queries in Python (select/where/join), and manages transactions with Sessions — the standard DB layer under FastAPI apps.
Overview
SQLAlchemy is Python's Hibernate: declarative models map classes to tables, a Session tracks changes and flushes them in transactions, and the 2.0 select() API composes type-checked queries. It prevents SQL injection by parameterizing everything, and its relationship() handles joins/foreign keys as object navigation. Learn the core loop — define models, open a session, add/query/commit — and you can build real backends; the same models plug directly into FastAPI dependencies.
Models, Session & CRUD
Mapped[type] columns define the schema; Session is a unit of work — objects you add or modify are written on commit(), rolled back on error.
from sqlalchemy import create_engine, String, select
from sqlalchemy.orm import (DeclarativeBase, Mapped, mapped_column,
Session)
class Base(DeclarativeBase): ...
class Student(Base):
__tablename__ = "students"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
branch: Mapped[str] = mapped_column(String(10))
cgpa: Mapped[float]
engine = create_engine("sqlite:///college.db", echo=False)
Base.metadata.create_all(engine) # CREATE TABLE
with Session(engine) as session:
session.add_all([
Student(name="Asha", branch="CS", cgpa=8.7),
Student(name="Ravi", branch="IT", cgpa=7.9),
])
session.commit() # one transaction
# Query — 2.0 style
stmt = (select(Student)
.where(Student.cgpa >= 8.0) # parameterized — no injection
.order_by(Student.cgpa.desc()))
for s in session.scalars(stmt):
print(s.name, s.cgpa) # Asha 8.7
# Update & delete are just object operations
asha = session.scalar(select(Student).where(Student.name == "Asha"))
asha.cgpa = 9.0
session.commit()Relationships — Foreign Keys as Attributes
relationship() lets you walk student.applications like a list while SQLAlchemy manages the JOINs. selectinload prevents the N+1 query problem — say those words in interviews.
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship, selectinload
class Company(Base):
__tablename__ = "companies"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str]
applications: Mapped[list["Application"]] = relationship(back_populates="company")
class Application(Base):
__tablename__ = "applications"
id: Mapped[int] = mapped_column(primary_key=True)
student_id: Mapped[int] = mapped_column(ForeignKey("students.id"))
company_id: Mapped[int] = mapped_column(ForeignKey("companies.id"))
status: Mapped[str] = mapped_column(default="applied")
company: Mapped["Company"] = relationship(back_populates="applications")
# Navigate objects instead of writing JOINs
# for app in student.applications: print(app.company.name, app.status)
# N+1 problem: looping students then touching .applications
# fires one query PER student. Fix — load eagerly:
stmt = select(Company).options(selectinload(Company.applications))
# Transactions roll back automatically on exceptions:
# with Session(engine) as s, s.begin():
# s.add(...) # commit on success, rollback on raiseKey Points to Remember
- 1Session = unit of work: add/modify objects, commit() writes one transaction
- 22.0 style: select(Model).where(...) with session.scalars() — parameterized, injection-safe
- 3relationship() navigates foreign keys as attributes; back_populates links both sides
- 4Know the N+1 problem and its fix (selectinload/joinedload) — a favourite interview probe
Interview Questions
Sign in to ask AriaWhat is an ORM? What do you gain and lose vs raw SQL?
Explain the N+1 query problem and how you would fix it in SQLAlchemy.
How does the Session manage transactions? What happens on an exception?
Ask Aria about SQLAlchemy — Python's Database Toolkit
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.