Using non-recursive CTEs to structure reporting queries

Break a reporting query into named stages with .cte() — scope, then aggregate, then rank — so the query reads in execution order, each stage can be selected on its own while debugging, and the scope filter exists in exactly one place. This guide belongs to common table expressions, CTEs and recursive queries.

Quick Answer

A report built from nested subqueries is read inside out, and the filter that defines "which orders count" ends up far from everything that depends on it.

Before — three levels of nesting:

from sqlalchemy import func, select

from shop.models import Customer, Order

per_customer = (
    select(Order.customer_id, func.sum(Order.total_cents).label("revenue"))
    .where(Order.placed_on >= since, Order.status != "cancelled")
    .group_by(Order.customer_id)
    .subquery()
)
ranked = (
    select(
        per_customer.c.customer_id,
        per_customer.c.revenue,
        func.rank().over(order_by=per_customer.c.revenue.desc()).label("position"),
    )
    .subquery()
)
stmt = (
    select(Customer.name, ranked.c.revenue, ranked.c.position)
    .join(Customer, Customer.id == ranked.c.customer_id)
    .where(ranked.c.position <= 20)
)

After — the same logic as named stages:

import datetime as dt

from sqlalchemy import func, select

from shop.models import Customer, Order


def top_customers(since: dt.date, limit: int = 20):
    # 1. Scope: the only place the reporting filter lives.
    orders_in_period = (
        select(Order.customer_id, Order.total_cents)
        .where(Order.placed_on >= since, Order.status != "cancelled")
        .cte("orders_in_period")
    )
    # 2. Aggregate.
    per_customer = (
        select(
            orders_in_period.c.customer_id,
            func.sum(orders_in_period.c.total_cents).label("revenue_cents"),
            func.count().label("order_count"),
        )
        .group_by(orders_in_period.c.customer_id)
        .cte("per_customer")
    )
    # 3. Rank.
    ranked = (
        select(
            per_customer,
            func.rank().over(order_by=per_customer.c.revenue_cents.desc()).label("position"),
        )
        .cte("ranked")
    )
    # 4. Label and present.
    return (
        select(Customer.name, ranked.c.revenue_cents, ranked.c.order_count, ranked.c.position)
        .join(Customer, Customer.id == ranked.c.customer_id)
        .where(ranked.c.position <= limit)
        .order_by(ranked.c.position)
    )

The rendered SQL is a WITH clause with three named stages, in the order they execute, and every stage can be run on its own by selecting from it.

Execution Context & Async Workflow Integration

A CTE is a named subquery declared in a WITH clause. select(...).cte("name") produces an object with a .c collection of its output columns, and referencing that object in another statement renders the WITH clause automatically — SQLAlchemy collects every CTE the final statement depends on, in dependency order.

Naming changes nothing about what the database computes, and a great deal about what a person can do with the query. A stage can be selected in isolation:

# Debugging: what does stage two actually produce?
rows = (await session.execute(select(per_customer).limit(5))).all()

That is the property worth optimising for in reporting code. A wrong total in a three-stage report is found by checking each stage's row count and sum against something known; in a nest of subqueries there is nothing to check.

On PostgreSQL 12 and later the planner decides whether to inline a CTE or materialise it. A CTE referenced once may be inlined, which lets predicates from the outer query push down into it — usually faster. A CTE referenced more than once is materialised: computed once, then reused, which is normally what a report wants, because the alternative is computing an expensive aggregate twice. Both can be forced when the planner chooses badly:

expensive = select(...).cte("expensive").prefix_with("MATERIALIZED")
cheap = select(...).cte("cheap").prefix_with("NOT MATERIALIZED")

Before PostgreSQL 12 every CTE was an optimisation fence — always materialised, never inlined — which is why older advice warns against CTEs in performance-sensitive queries. On current versions that advice is obsolete for the single-reference case.

Under async this is all statement construction, so the whole report is built synchronously and executed with one await session.execute(stmt). Two async-specific points are worth keeping in mind. Reports return rows rather than entities, so there are no relationships to lazy-load and nothing to raise MissingGreenlet — which is one reason column-based reporting is comfortable in async code. And a large report should stream rather than buffer, which yield_per handles, as in using yield_per to stream millions of rows.

Resolving Warnings, Errors & Common Mistakes

The fan-out mistake is the one that produces confidently wrong numbers. Joining orders to order lines and then summing orders.total_cents counts each order once per line:

# WRONG: total_cents is repeated per line, so revenue is multiplied by lines per order.
bad = (
    select(func.sum(Order.total_cents))
    .join(OrderLine, OrderLine.order_id == Order.id)
    .where(OrderLine.product_id == product_id)
)

# RIGHT: identify the orders first, then sum each one once.
matching_orders = (
    select(OrderLine.order_id)
    .where(OrderLine.product_id == product_id)
    .distinct()
    .cte("matching_orders")
)
good = (
    select(func.sum(Order.total_cents))
    .join(matching_orders, matching_orders.c.order_id == Order.id)
)

Staging makes the fix natural: the first stage answers "which orders", the second answers "how much", and neither can accidentally do both. It is the same reasoning behind using EXISTS instead of a join when filtering by a collection, described in fixing cartesian product warnings in SQLAlchemy joins.

Advanced: Reusing Stages, Parameters and Composition

Because a CTE is an ordinary Python object, reporting queries compose like code. A function returning a stage can be reused across reports, and the scope filter can be passed in rather than repeated:

import datetime as dt

from sqlalchemy import Select, func, select

from shop.models import Order


def order_scope(since: dt.date, until: dt.date, tenant_id: int):
    """The single definition of 'orders that count' for every report."""
    return (
        select(Order.id, Order.customer_id, Order.placed_on, Order.total_cents)
        .where(
            Order.tenant_id == tenant_id,
            Order.placed_on >= since,
            Order.placed_on < until,
            Order.status.not_in(("cancelled", "test")),
        )
        .cte("order_scope")
    )


def revenue_by_day(since: dt.date, until: dt.date, tenant_id: int) -> Select:
    scope = order_scope(since, until, tenant_id)
    return (
        select(
            scope.c.placed_on.label("day"),
            func.sum(scope.c.total_cents).label("revenue_cents"),
            func.count().label("order_count"),
        )
        .group_by(scope.c.placed_on)
        .order_by(scope.c.placed_on)
    )


def revenue_by_customer(since: dt.date, until: dt.date, tenant_id: int) -> Select:
    scope = order_scope(since, until, tenant_id)
    return (
        select(scope.c.customer_id, func.sum(scope.c.total_cents).label("revenue_cents"))
        .group_by(scope.c.customer_id)
        .order_by(func.sum(scope.c.total_cents).desc())
    )

Two reports, one definition of which orders count. When the definition changes — a new status to exclude, a tenant rule — it changes in one function, and no report can be left behind. That is the real argument for CTE-based reporting over hand-written SQL strings: composition is ordinary Python.

Referencing one stage twice in the same statement is where materialisation matters. A comparison of each customer's revenue against the overall average needs the aggregate stage twice, once grouped and once as a scalar:

per_customer = (
    select(scope.c.customer_id, func.sum(scope.c.total_cents).label("revenue_cents"))
    .group_by(scope.c.customer_id)
    .cte("per_customer")
)
average = select(func.avg(per_customer.c.revenue_cents)).scalar_subquery()

stmt = (
    select(
        per_customer.c.customer_id,
        per_customer.c.revenue_cents,
        (per_customer.c.revenue_cents - average).label("vs_average_cents"),
    )
    .order_by(per_customer.c.revenue_cents.desc())
)

PostgreSQL materialises per_customer because it is referenced more than once, so the aggregate runs once rather than twice — exactly the behaviour a report wants, and one that a nested-subquery version would not have expressed as clearly.

For the window-function side of reporting — running totals, period-over-period change, top N per group — the same staging applies, with the window function in a later stage over the aggregate. Those are covered in window functions and analytical queries.

Testing a Multi-Stage Report

Reports are hard to test because the correct answer is usually "whatever the data says". The way out is to test each stage against a fixture whose totals you computed by hand, and to assert the shape of each stage as well as the final numbers.

import datetime as dt

import pytest
from sqlalchemy import func, select

from shop.reports import order_scope, revenue_by_customer

SINCE = dt.date(2026, 9, 1)
UNTIL = dt.date(2026, 10, 1)


@pytest.fixture
async def report_data(session, order_factory):
    # Two customers, one cancelled order, one order outside the period.
    await order_factory(customer_id=1, placed_on=dt.date(2026, 9, 5), total_cents=1000)
    await order_factory(customer_id=1, placed_on=dt.date(2026, 9, 6), total_cents=2500)
    await order_factory(customer_id=2, placed_on=dt.date(2026, 9, 7), total_cents=700)
    await order_factory(customer_id=1, placed_on=dt.date(2026, 9, 8), total_cents=9999,
                        status="cancelled")
    await order_factory(customer_id=1, placed_on=dt.date(2026, 8, 31), total_cents=9999)
    await session.commit()


@pytest.mark.asyncio
async def test_scope_excludes_cancelled_and_out_of_period(session, report_data):
    scope = order_scope(SINCE, UNTIL, tenant_id=1)
    count = await session.scalar(select(func.count()).select_from(scope))
    assert count == 3          # the cancelled and the August order are excluded


@pytest.mark.asyncio
async def test_revenue_by_customer(session, report_data):
    rows = (await session.execute(revenue_by_customer(SINCE, UNTIL, tenant_id=1))).all()
    assert rows == [(1, 3500), (2, 700)]

The first test is the one that pays off repeatedly. It asserts the scope stage in isolation, so a change to the status exclusions or the period boundaries fails with a message about the scope rather than about a revenue figure three stages later. Boundary rows — the order on the first day of the period, the order on the last day, the cancelled one — are what make it meaningful.

Two more assertions are worth adding to any report that joins. Assert the row count of the aggregate stage equals the number of distinct group keys in the fixture, which catches fan-out immediately. And assert that the sum of the per-group values equals the overall total computed independently, which catches rows being dropped by an inner join that should have been outer.

Run these against PostgreSQL rather than SQLite: window functions, FILTER, date_trunc and generate_series behave differently or do not exist elsewhere, and a report tested on SQLite is not tested. The container fixture is the one from running tests against a Postgres testcontainer.

Frequently Asked Questions

Related

• Common Table Expressions, CTEs and Recursive Queries — The parent guide: CTEs for decomposition and recursion.
• Using CTEs with INSERT, UPDATE and DELETE ... RETURNING — Stages that write rather than read.
• Selecting the top N rows per group with row_number() — Window functions in a later stage of the same shape.
• Fixing cartesian product warnings in SQLAlchemy joins — The fan-out that inflates report totals.