SQLAlchemy Core Ideas Checkpoint
25 min · Intermediate
Need help? Ask a tutor on WhatsApp (opens in a new tab)Fix the mistakes
The slips that cost marks in this module. Spot each one, then check the fix.
Real code keeps database work in small functions. They take a session, run a statement and return plain results. This checkpoint builds such a repository and then tests the whole module.
One engine, one unit of work
Create the engine once. Wrap the session in a small helper so every job gets a new session, a commit on success and a rollback on error.
@contextmanagerdef unit_of_work(): with Session(engine) as session, session.begin(): yield session
with unit_of_work() as session: add_product(session, "Pen", "stationery", 10)Example 1 · A repository working in one transaction
from contextlib import contextmanagerfrom sqlalchemy import create_engine, select, func, String, CheckConstraintfrom sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class Base(DeclarativeBase): pass
class Product(Base): __tablename__ = "product" __table_args__ = (CheckConstraint("price >= 0"),) id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(50), unique=True) category: Mapped[str] = mapped_column(String(30)) price: Mapped[int] stock: Mapped[int] = mapped_column(default=0)
engine = create_engine("sqlite://")Base.metadata.create_all(engine)
@contextmanagerdef unit_of_work(): with Session(engine) as session, session.begin(): yield session
def add_product(session, name, category, price, stock=0): product = Product(name=name, category=category, price=price, stock=stock) session.add(product) return product
def by_category(session, category): stmt = select(Product.name).where(Product.category == category).order_by(Product.name) return session.scalars(stmt).all()
with unit_of_work() as session: add_product(session, "Pen", "stationery", 10) add_product(session, "Notebook", "stationery", 60) add_product(session, "Desk", "furniture", 7000)
with unit_of_work() as session: print(by_category(session, "stationery")) print(by_category(session, "toys")) print(session.get(Product, 3).name, session.get(Product, 9))['Notebook', 'Pen'] [] Desk None
What this code does
- The three products are saved together when the first block ends.
- An unknown category gives an empty list and an unknown id gives
None. Neither is an error.
The caller never writes commit or rollback. The helper does it, so no job forgets.
Small functions take the session
A function that receives a session is easy to test and reuse. It builds a statement, runs it and returns data. It does not create sessions or commit.
def by_category(session, category): stmt = ( select(Product.name) .where(Product.category == category) .order_by(Product.name) ) return session.scalars(stmt).all()The caller decides when to commit. This keeps one job as one transaction, even if it calls several functions.
Pick the result method by the answer you expect
Ask what shape of answer you need. A list of objects, a single value, one optional object or rows of several columns each have their own method.
names = session.scalars(select(Product.name)).all()total = session.scalar(select(func.count()).select_from(Product))product = session.get(Product, 1)rows = session.execute(select(Product.name, Product.price)).all()Example 2 · A failed job leaves nothing behind
from contextlib import contextmanagerfrom sqlalchemy import create_engine, select, func, String, CheckConstraintfrom sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class Base(DeclarativeBase): pass
class Product(Base): __tablename__ = "product" __table_args__ = (CheckConstraint("price >= 0"),) id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(50), unique=True) category: Mapped[str] = mapped_column(String(30)) price: Mapped[int] stock: Mapped[int] = mapped_column(default=0)
engine = create_engine("sqlite://")Base.metadata.create_all(engine)
@contextmanagerdef unit_of_work(): with Session(engine) as session, session.begin(): yield session
def add_product(session, name, category, price, stock=0): product = Product(name=name, category=category, price=price, stock=stock) session.add(product) return product
def by_category(session, category): stmt = select(Product.name).where(Product.category == category).order_by(Product.name) return session.scalars(stmt).all()
from sqlalchemy.exc import IntegrityError
with unit_of_work() as session: add_product(session, "Pen", "stationery", 10)
try: with unit_of_work() as session: add_product(session, "Ink", "stationery", 20) add_product(session, "Pen", "stationery", 30)except IntegrityError: print("job failed")
with unit_of_work() as session: print(session.scalars(select(Product.name)).all())job failed ['Pen']
What this code does
- The second job adds Ink and then a second Pen, which breaks the unique name.
- The whole job rolled back, so Ink was not saved either.
get finds by primary key and returns None when missing. scalar gives one value. execute keeps several columns together.
Finish up
Mark complete & continue
Next: One-to-Many and Many-to-Many