Databases with an ORM

Relationships

A foreign key links rows; a relationship turns that link into Python attributes on both sides:

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    reviews: Mapped[list["Review"]] = relationship(back_populates="user")

class Review(Base):
    __tablename__ = "reviews"
    id: Mapped[int] = mapped_column(primary_key=True)
    stars: Mapped[int]
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    user: Mapped["User"] = relationship(back_populates="reviews")

Now review.user is the User object and user.reviews is a list of their reviews. Appending a new review to user.reviews sets its user_id for you when the session commits. back_populates keeps both sides in step.

By default a relationship is lazy: user.reviews isn't loaded until you first touch it, and then it runs a query. That's convenient, and it's also the source of the most common ORM performance bug. It can also surprise you after the session closes: touching a relationship that was never loaded then raises an error, because there's no session left to run the query.

The N+1 problem

Loop over 100 users and print each one's review count. That's 1 query for the users, plus 1 more each time the loop touches user.reviews: 101 queries, the N+1 problem. It's invisible in a small test and slow in production. Count the queries to see it, then ask for the related rows up front with selectinload:

from sqlalchemy import ForeignKey, create_engine, event, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship, selectinload

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    reviews: Mapped[list["Review"]] = relationship(back_populates="user")

class Review(Base):
    __tablename__ = "reviews"
    id: Mapped[int] = mapped_column(primary_key=True)
    stars: Mapped[int]
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    user: Mapped[User] = relationship(back_populates="reviews")

engine = create_engine("sqlite://")
Base.metadata.create_all(engine)
queries = []
event.listen(engine, "before_cursor_execute", lambda *args: queries.append(args[2]))

with Session(engine) as s:
    for i in range(20):
        s.add(User(name=f"member{i}", reviews=[Review(stars=(i % 5) + 1), Review(stars=4)]))
    s.commit()

with Session(engine) as s:
    queries.clear()
    counts = [len(u.reviews) for u in s.scalars(select(User))]
    print("lazy:", len(queries), "queries")

with Session(engine) as s:
    queries.clear()
    counts = [len(u.reviews) for u in s.scalars(select(User).options(selectinload(User.reviews)))]
    print("selectinload:", len(queries), "queries")

selectinload(User.reviews) loads every user's reviews in one extra query (WHERE user_id IN (...)), so the total is 2 however many users there are. Counting queries with an event listener, or turning on create_engine(..., echo=True) to print the SQL, is how you catch this before it reaches production.

from sqlalchemy import ForeignKey, create_engine, func, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

class Base(DeclarativeBase):
    pass

class Product(Base):
    __tablename__ = "products"
    id: Mapped[int] = mapped_column(primary_key=True)
    sku: Mapped[str]

class Review(Base):
    __tablename__ = "reviews"
    id: Mapped[int] = mapped_column(primary_key=True)
    product_id: Mapped[int] = mapped_column(ForeignKey("products.id"))
    stars: Mapped[int]

engine = create_engine("sqlite://")
Base.metadata.create_all(engine)
with Session(engine) as s:
    s.add_all([Product(id=1, sku="TE-101"), Product(id=2, sku="PA-108"), Product(id=3, sku="CA-132")])
    s.add_all([Review(product_id=p, stars=st) for p, st in [(1, 5), (1, 4), (1, 3), (2, 5), (2, 5), (3, 5)]])
    s.commit()
    ranking = []
    print(ranking)

Transactions, and when to write SQL yourself

A session groups changes into a transaction: commit() saves them all, rollback() throws them all away. If anything raises before the commit, nothing is half-saved. with session.begin(): commits at the end of the block, or rolls back if the block raises, which is the safe pattern for a group of changes that must happen together, like moving stock between stores.

The ORM is great for the everyday work of loading objects, following relationships and saving changes. For reports full of aggregates, window functions or database-specific features, plain SQL is often shorter and clearer. SQLAlchemy lets you write it with text(), and still binds parameters safely:

from sqlalchemy import text
rows = session.execute(text("SELECT sku, price FROM products WHERE price < :limit ORDER BY price"), {"limit": 50}).all()

Never build SQL by pasting values into the string with an f-string. That's how SQL injection happens, as the security lesson showed. Both select() and text() with :name placeholders send values separately from the query.

In a real app, the schema changes over time. A tool called Alembic records each change as a migration script, so every copy of the database can be upgraded the same way.

Keep sessions short. A web app typically opens one session per request and closes it when the response is sent; a script opens one per unit of work. A session held open for hours keeps stale copies of rows in memory and can hold database locks that block everyone else. Sessions aren't safe to share between threads either: give each thread or request its own, all made from one shared engine. The engine is the long-lived object; it keeps a pool of connections ready, so opening a session is cheap.

from sqlalchemy import CheckConstraint, String, create_engine, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

class Base(DeclarativeBase):
    pass

class Stock(Base):
    __tablename__ = "stock"
    __table_args__ = (CheckConstraint("qty >= 0", name="no_negative_stock"),)
    id: Mapped[int] = mapped_column(primary_key=True)
    store: Mapped[str] = mapped_column(String(20))
    sku: Mapped[str] = mapped_column(String(6))
    qty: Mapped[int]

engine = create_engine("sqlite://")
Base.metadata.create_all(engine)
with Session(engine) as s:
    s.add_all([Stock(store="portland", sku="TE-101", qty=5), Stock(store="denver", sku="TE-101", qty=1)])
    s.commit()

def move(sku, src, dst, qty):
    with Session(engine) as session, session.begin():
        rows = {r.store: r for r in session.scalars(select(Stock).where(Stock.sku == sku))}
        rows[dst].qty += qty
        rows[src].qty -= qty

def levels():
    with Session(engine) as session:
        return {r.store: r.qty for r in session.scalars(select(Stock))}

move("TE-101", "portland", "denver", 2)
print("after a good move:", levels())
try:
    move("TE-101", "denver", "portland", 10)
except IntegrityError:
    print("refused: denver doesn't have 10")
print("after the refused move:", levels())