TopolearnReports like 'latest order per customer' or running totals need subqueries and windows, and SQLAlchemy builds them from reusable pieces.
Any select() can become a building block:
.subquery()turns it into a derived table you can join to or select from. Its columns live under.c, like a Table..cte("name")does the same as aWITHclause, which reads better for multi-step queries..cte(recursive=True)plusunion_allbuilds recursive CTEs for trees..scalar_subquery()makes a one-value subquery usable as a column or in a comparison, e.g. a correlated count per row. It correlates automatically with the enclosing query's tables.
Window functions use .over() on a function: func.row_number().over(partition_by=Order.customer_id, order_by=Order.created.desc()), or func.sum(Order.total).over(order_by=Order.created) for a running total. Windows can't appear in WHERE, so the classic "top 1 per group" is a window in a subquery or CTE, filtered outside.
To get ORM objects back from a subquery, use aliased(Order, subq), which maps the entity onto the subquery's columns. Then select(alias) returns Order instances.
Build complex queries in named steps, print them to check the SQL, and remember each step is immutable and reusable.
Example
from sqlalchemy import create_engine, select, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class Base(DeclarativeBase): pass
class Order(Base):
__tablename__ = "orders"
id: Mapped[int] = mapped_column(primary_key=True)
customer: Mapped[str]
total: Mapped[int]
engine = create_engine("sqlite://"); Base.metadata.create_all(engine)
with Session(engine) as s:
s.add_all([Order(customer=c, total=t) for c, t in
[("ada", 30), ("ada", 50), ("bob", 20), ("bob", 70), ("bob", 10)]])
s.commit()
ranked = select(Order.customer, Order.total,
func.row_number().over(partition_by=Order.customer,
order_by=Order.total.desc()).label("rn")
).cte("ranked")
top = select(ranked.c.customer, ranked.c.total).where(ranked.c.rn == 1)
print(s.execute(top.order_by(ranked.c.customer)).all()) # [('ada', 50), ('bob', 70)]
avg = select(func.avg(Order.total)).scalar_subquery()
print(s.scalars(select(Order.id).where(Order.total > avg)).all()) # [2, 4]Remember
- subquery() and cte() turn a select into a table-like object with .c
- scalar_subquery() gives a single value usable in expressions
- func.x().over(partition_by=..., order_by=...) for window functions
- aliased(Model, subq) maps ORM entities onto a subquery
Watch out for
- Trying to filter on a window function in the same query's WHERE clause.