Selecting the top N rows per group with row_number()

Compute func.row_number().over(partition_by=Order.customer_id, order_by=Order.placed_at.desc()) in a subquery, filter the row number in an outer query, and map the result back with aliased(Order, subquery) — window functions cannot be filtered in the WHERE clause of the query that computes them. This guide belongs to window functions and analytical queries.

Quick Answer

The instinctive version puts the window function in WHERE and fails; the Python version loads everything and slices. One subquery fixes both.

Before — filtering a window function in place, or slicing in Python:

from sqlalchemy import func, select

from shop.models import Order

rn = func.row_number().over(partition_by=Order.customer_id, order_by=Order.placed_at.desc())
stmt = select(Order).where(rn <= 3)
# ProgrammingError: window functions are not allowed in WHERE


async def latest_three_per_customer_slow(session, customer_ids: list[int]):
    orders = (await session.scalars(
        select(Order).where(Order.customer_id.in_(customer_ids)).order_by(Order.placed_at.desc())
    )).all()
    grouped: dict[int, list[Order]] = {}
    for order in orders:                         # loads every order of every customer
        grouped.setdefault(order.customer_id, [])
        if len(grouped[order.customer_id]) < 3:
            grouped[order.customer_id].append(order)
    return grouped

After — number in a subquery, filter outside, map back to Order:

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import aliased

from shop.models import Order


async def latest_three_per_customer(
    session: AsyncSession, customer_ids: list[int], n: int = 3
) -> list[Order]:
    numbered = (
        select(
            Order,
            func.row_number()
            .over(
                partition_by=Order.customer_id,
                order_by=(Order.placed_at.desc(), Order.id.desc()),
            )
            .label("rn"),
        )
        .where(Order.customer_id.in_(customer_ids))
        .subquery()
    )
    recent = aliased(Order, numbered)
    stmt = (
        select(recent)
        .where(numbered.c.rn <= n)
        .order_by(recent.customer_id, numbered.c.rn)
    )
    return list(await session.scalars(stmt))

Order.id.desc() in the window ordering is a tiebreaker: two orders placed in the same second would otherwise be numbered in an arbitrary order that can change between runs.

Execution Context & Async Workflow Integration

SQL evaluates a query in a fixed logical order: FROM and joins, then WHERE, then GROUP BY and HAVING, then window functions, then SELECT's output, ORDER BY and LIMIT. Window functions are computed from the rows that survived WHERE, so WHERE cannot refer to their results — they do not exist yet. PostgreSQL says so directly: window functions are not allowed in WHERE. Wrapping the query in a subquery moves the window results into an inner level, where the outer WHERE can filter them like any column.

over(partition_by=..., order_by=...) maps onto OVER (PARTITION BY ... ORDER BY ...). The partition restarts numbering for each customer; the ordering decides which order is number one. The .label("rn") names the column so the outer query can reach it as numbered.c.rn.

aliased(Order, numbered) tells the ORM that the subquery's columns have the shape of an Order. Selecting recent then produces real Order objects, placed in the identity map like any other loaded row. The extra rn column is present in the subquery but not part of the entity, so it is simply ignored unless you select it alongside: select(recent, numbered.c.rn) returns (Order, int) rows.

Under async, nothing about the construction changes; await session.scalars(stmt) runs it in one round trip. The async-specific concern is what happens next. Relationships on the returned orders are not loaded, and touching order.lines lazily raises MissingGreenlet. Add loader options to the outer select — select(recent).options(selectinload(recent.lines)) — so the lines for exactly the top-N orders are fetched in one additional query. Loader options on an aliased entity use the alias's attributes, not Order's. The broader loading guidance is in using selectinload vs joinedload for N+1 prevention.

The same shape works in Core for reports that need no objects: select columns instead of the entity, and read rows. The parent topic on window functions and analytical queries covers frames and running totals, which use the same over() construct.

Resolving Warnings, Errors & Common Mistakes

The cartesian-product warning is worth a closer look, because the query it produces is wrong and often huge. Writing select(Order).where(numbered.c.rn <= 3) puts two independent sources in the FROM clause — the orders table and the subquery — with no join condition between them, so every order is paired with every numbered row. SQLAlchemy 1.4 and later warn about exactly this. The fix is to select from the subquery, which is what aliased() expresses, as explained in fixing cartesian product warnings in SQLAlchemy joins.

Choosing among the numbering functions is about ties. row_number() always yields exactly N rows per group, breaking ties arbitrarily unless you give it a tiebreaker. rank() gives tied rows the same number and skips the following numbers, so "top 2" keeps both tied rows and nothing after them. dense_rank() gives tied rows the same number without gaps, so "top 2" can return three or more rows. "The latest three orders" is a row_number() question; "the products in the top two price points" is a dense_rank() question.

Advanced: LATERAL, DISTINCT ON and Row-Limited Relationships

The window-function version numbers every row in each partition before discarding most of them. When groups are small that is irrelevant. When each customer has thousands of orders and you want three, a LATERAL subquery with LIMIT reads only what it needs, provided an index on (customer_id, placed_at) exists.

from sqlalchemy import Index, select, true
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import aliased

from shop.models import Customer, Order

# In the model: Index("ix_orders_customer_id_placed_at", Order.customer_id, Order.placed_at.desc())


async def latest_three_lateral(session: AsyncSession, customer_ids: list[int]):
    latest = (
        select(Order)
        .where(Order.customer_id == Customer.id)
        .order_by(Order.placed_at.desc(), Order.id.desc())
        .limit(3)
        .lateral()
    )
    recent = aliased(Order, latest)
    stmt = (
        select(Customer, recent)
        .join(recent, true())
        .where(Customer.id.in_(customer_ids))
        .order_by(Customer.id, recent.placed_at.desc())
    )
    return (await session.execute(stmt)).all()

For each customer, PostgreSQL runs the inner query as an index scan that stops after three entries. The join condition lives inside the lateral subquery, so the outer join(..., true()) needs no further condition. Use isouter=True to keep customers with no orders.

When N is one, PostgreSQL's DISTINCT ON is shorter still:

from sqlalchemy import select

from shop.models import Order

latest_per_customer = (
    select(Order)
    .distinct(Order.customer_id)
    .order_by(Order.customer_id, Order.placed_at.desc(), Order.id.desc())
)

The ORDER BY must begin with the DISTINCT ON expressions; the rest of the ordering picks which row survives.

Finally, when "the latest three orders" is something the application asks for everywhere, it can be a relationship. A viewonly relationship to an aliased, window-filtered selectable lets selectinload(Customer.recent_orders) load exactly three per customer:

from sqlalchemy import func, select
from sqlalchemy.orm import aliased, relationship

from shop.models import Customer, Order

numbered = select(
    Order,
    func.row_number().over(
        partition_by=Order.customer_id, order_by=(Order.placed_at.desc(), Order.id.desc())
    ).label("rn"),
).subquery()
RecentOrder = aliased(Order, numbered)

Customer.recent_orders = relationship(
    RecentOrder,
    primaryjoin=(Customer.id == RecentOrder.customer_id) & (numbered.c.rn <= 3),
    order_by=RecentOrder.placed_at.desc(),
    viewonly=True,
)

It is assigned after both classes exist because it needs the mapped Order. Being viewonly, it cannot be appended to; writes still go through Customer.orders.

Top N by an Aggregate: Best Sellers per Category

Many top-N questions rank groups by something computed, not by a stored column: the five best-selling products in each category last month, the three customers with the highest spend per region. The pattern gains one step — aggregate first, then number the aggregates — and a CTE keeps each step readable.

import datetime as dt

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import OrderLine, Product


async def best_sellers_per_category(
    session: AsyncSession, since: dt.date, n: int = 5
) -> list[dict]:
    # 1. Aggregate: units sold per product since the cutoff.
    sales = (
        select(
            Product.category_id,
            Product.id.label("product_id"),
            Product.name,
            func.sum(OrderLine.quantity).label("units"),
        )
        .join(OrderLine, OrderLine.product_id == Product.id)
        .where(OrderLine.created_at >= since)
        .group_by(Product.category_id, Product.id, Product.name)
        .cte("sales")
    )
    # 2. Number the aggregates within each category.
    ranked = (
        select(
            sales,
            func.rank()
            .over(partition_by=sales.c.category_id, order_by=sales.c.units.desc())
            .label("position"),
        )
        .cte("ranked")
    )
    # 3. Keep the top n, ties included.
    stmt = (
        select(ranked)
        .where(ranked.c.position <= n)
        .order_by(ranked.c.category_id, ranked.c.position, ranked.c.name)
    )
    return [dict(row._mapping) for row in await session.execute(stmt)]

rank() is the deliberate choice here. Two products that sold exactly the same number of units are equally "top five", and a merchandising report that silently dropped one of them would be wrong; rank() keeps both and may return six rows for a category. If the consumer needs exactly five — a fixed-size widget — switch to row_number() and add the product name or id as a tiebreaker so the choice is stable.

The window function can also sit directly over the aggregate in a single select, because window functions are evaluated after GROUP BY: func.rank().over(partition_by=Product.category_id, order_by=func.sum(OrderLine.quantity).desc()) is valid in the same query that groups. The outer filter still needs a second level, so the CTE version is no longer, and it is much easier to debug — each CTE can be selected on its own to check its numbers.

For dashboards that run this constantly, the aggregation is the expensive part, not the ranking. A materialised view refreshed on a schedule, or a summary table maintained by the order pipeline, turns the query into a cheap window function over a few thousand pre-aggregated rows. The CTE patterns for insert, update and delete show how to maintain such a summary in the same transaction as the writes that change it.

Frequently Asked Questions

Related

• Window Functions and Analytical Queries — The parent guide: partitions, frames and analytical patterns.
• Writing window functions for running totals in Python — Frames and cumulative sums with the same over() construct.
• Fixing cartesian product warnings in SQLAlchemy joins — Why selecting from both a table and its subquery goes wrong.
• Paginating large result sets with keyset pagination — The same composite index, used for paging instead.