Writing Window Functions for Running Totals in Python

Use func.sum(col).over(partition_by=..., order_by=..., rows=(None, 0)) in a SQLAlchemy 2.0 select() statement — rows=(None, 0) compiles to ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW, which guarantees physical row ordering and avoids the quadratic tie-breaking behaviour of the default RANGE frame; this pattern and its async execution model are covered in the parent Window Functions and Analytical Queries guide.

Quick Answer

The minimal before/after replacement from legacy ORM to modern 2.0:

# Sync — SQLAlchemy 1.x legacy (do not use in 2.0 codebases)
from sqlalchemy.orm import Session
from sqlalchemy import func

def running_total_legacy(session: Session):
    return (
        session.query(
            Transaction.id,
            Transaction.amount,
            func.sum(Transaction.amount)
            .over(order_by=Transaction.created_at)
            .label("running_total"),
        )
        .all()
    )

# Async — SQLAlchemy 2.0 canonical form with frame clause
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped
from datetime import datetime
from decimal import Decimal


class Base(DeclarativeBase):
    pass


class Transaction(Base):
    __tablename__ = "transactions"

    id: Mapped[int] = mapped_column(primary_key=True)
    account_id: Mapped[int] = mapped_column(nullable=False)
    amount: Mapped[Decimal] = mapped_column(nullable=False)
    created_at: Mapped[datetime] = mapped_column(nullable=False)


async def running_total_2x(session: AsyncSession) -> list:
    stmt = select(
        Transaction.id,
        Transaction.account_id,
        Transaction.amount,
        Transaction.created_at,
        func.sum(Transaction.amount)
        .over(
            partition_by=Transaction.account_id,
            order_by=Transaction.created_at,
            rows=(None, 0),  # ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        )
        .label("running_total"),
    ).order_by(Transaction.account_id, Transaction.created_at)

    result = await session.execute(stmt)
    return result.all()

Key differences: session.query() is removed in 2.0; select() is universal — the same shift you apply when migrating legacy 1.4 code to 2.0 syntax across a codebase. await session.execute() replaces synchronous .all() on the query object. The rows=(None, 0) argument is new — it changes the generated OVER clause from the implicit RANGE UNBOUNDED PRECEDING AND CURRENT ROW to the more performant ROWS UNBOUNDED PRECEDING AND CURRENT ROW.

Execution Context & Async Workflow Integration

When await session.execute(stmt) runs, asyncpg sends the compiled SQL to PostgreSQL and opens a cursor. The database evaluates the window function server-side: it sorts the rows within each partition by created_at, then walks forward computing the cumulative sum. No Python-side arithmetic happens.

The result object returned by await session.execute() is a CursorResult. Calling .all() on it buffers every row in Python memory and closes the cursor. For small datasets this is fine. For tables with millions of rows, stream the result set with yield_per instead of buffering:

# Async — streaming a large running total result set
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from typing import AsyncIterator


async def stream_running_totals(
    session: AsyncSession,
    chunk_size: int = 500,
) -> AsyncIterator[list]:
    stmt = select(
        Transaction.account_id,
        Transaction.created_at,
        Transaction.amount,
        func.sum(Transaction.amount)
        .over(
            partition_by=Transaction.account_id,
            order_by=Transaction.created_at,
            rows=(None, 0),
        )
        .label("running_total"),
    ).execution_options(yield_per=chunk_size)

    result = await session.execute(stmt)
    try:
        async for batch in result.partitions(chunk_size):
            yield batch
    finally:
        await result.close()

The try/finally around the iterator is important: if the caller cancels the coroutine or raises before exhausting the stream, result.close() releases the server-side cursor and returns the connection to the pool. Without it, the connection remains checked out until garbage collection, which under load causes pool exhaustion.

Connection pool interaction: each await session.execute() call acquires a connection from the pool for the duration of result consumption. With streaming (yield_per), the connection stays checked out across every iteration of the async loop. Size your pool (pool_size, max_overflow) to account for concurrent streaming queries — a single streaming consumer holds a connection for the entire duration of iteration, and getting this wrong is a leading cause of connection leaks and pool exhaustion.

One important property of running total queries in async environments: because await yields control back to the event loop between iterations of an async generator, other coroutines can run concurrently. This means a second coroutine could commit a write to the transactions table while your streaming query is mid-result. If you need a consistent snapshot across the entire result, wrap the streaming operation in an explicit REPEATABLE READ transaction started before the first execute() call — see transaction isolation and commit strategies for how to set the isolation level per transaction. Under the default READ COMMITTED isolation, each row-fetch within the stream may see different committed data, which produces an inconsistent running total.

Result materialization format matters for downstream consumers. Use .all() for tuple access, .mappings().all() for dict-like access by column name, or chain .scalars() only when the query projects a single column. For running totals — which always project multiple columns — .scalars() silently discards all columns except the first, a common source of bugs when migrating from single-column queries.

# Async — engine tuned for concurrent analytical streaming
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@db/analytics",
    pool_size=15,
    max_overflow=5,
    pool_timeout=30,
    pool_pre_ping=True,
    pool_recycle=3600,
)

Resolving Warnings, Errors & Common Mistakes

Advanced Running Total Optimization

Incremental Materialization with INSERT ... SELECT

For reporting tables refreshed on a schedule, avoid recomputing the full window over all historical rows. Instead, compute running totals only over new rows since the last refresh and append them with an insert().from_select() bulk write that keeps the whole refresh in a single round trip:

# Async — incremental running total refresh
from sqlalchemy import func, select, insert, text, Table, Column, Integer, Numeric, DateTime, MetaData
from sqlalchemy.ext.asyncio import AsyncSession
from datetime import datetime, timezone

metadata = MetaData()

RunningTotals = Table(
    "running_totals_cache",
    metadata,
    Column("account_id", Integer, nullable=False),
    Column("as_of", DateTime(timezone=True), nullable=False),
    Column("running_total", Numeric(18, 4), nullable=False),
)


async def incremental_refresh(session: AsyncSession, since: datetime) -> None:
    # Find the last cached total per account to use as baseline
    last_cached = (
        select(
            RunningTotals.c.account_id,
            func.max(RunningTotals.c.as_of).label("last_refresh"),
            func.max(RunningTotals.c.running_total).label("baseline"),
        )
        .group_by(RunningTotals.c.account_id)
        .cte("last_cached")
    )

    # Compute incremental window over new rows only
    new_rows = select(
        Transaction.account_id,
        Transaction.created_at,
        (
            func.coalesce(last_cached.c.baseline, Decimal("0"))
            + func.sum(Transaction.amount).over(
                partition_by=Transaction.account_id,
                order_by=Transaction.created_at,
                rows=(None, 0),
            )
        ).label("running_total"),
    ).join(
        last_cached,
        Transaction.account_id == last_cached.c.account_id,
        isouter=True,
    ).where(
        Transaction.created_at > since
    ).cte("incremental")

    async with session.begin():
        await session.execute(
            insert(RunningTotals).from_select(
                ["account_id", "as_of", "running_total"],
                select(
                    new_rows.c.account_id,
                    new_rows.c.created_at,
                    new_rows.c.running_total,
                ),
            )
        )

This pattern reduces computation from O(all rows) to O(new rows since last refresh), which is critical for accounts with years of historical data. The incremental approach relies on monotonically increasing created_at values — if rows can be backdated, a full recompute is safer.

Using func.sum().filter() for Conditional Running Totals

When you need a running total that only accumulates certain rows — for example, credits but not debits, or confirmed orders but not cancelled ones — use the FILTER (WHERE ...) aggregate syntax rather than post-processing in Python:

# Async — running total of credits only, debits excluded
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession


async def credit_running_total(session: AsyncSession) -> list:
    stmt = select(
        Transaction.account_id,
        Transaction.created_at,
        Transaction.amount,
        func.sum(Transaction.amount)
        .filter(Transaction.amount > 0)  # credits only
        .over(
            partition_by=Transaction.account_id,
            order_by=Transaction.created_at,
            rows=(None, 0),
        )
        .label("credit_running_total"),
        func.sum(Transaction.amount)
        .filter(Transaction.amount < 0)  # debits only
        .over(
            partition_by=Transaction.account_id,
            order_by=Transaction.created_at,
            rows=(None, 0),
        )
        .label("debit_running_total"),
    )
    result = await session.execute(stmt)
    return result.mappings().all()

This compiles to SUM(amount) FILTER (WHERE amount > 0) OVER (...) — a single database pass that produces both columns simultaneously. The alternative — two separate queries or Python-side filtering — doubles I/O or shifts work off the database, both of which are worse under load.

Resetting a Running Total at a Boundary

A less obvious but frequently requested pattern is resetting the running total at a logical boundary within a partition — for example, restarting the cumulative sum at the beginning of each calendar month while still partitioning by account. SQL does not have a native "reset on value change" frame clause, but you can solve this cleanly by composing the window function on a group-keyed expression rather than a raw timestamp column.

# Async — running total that resets at the start of each calendar month
from sqlalchemy import func, select, extract
from sqlalchemy.ext.asyncio import AsyncSession


async def monthly_running_totals(session: AsyncSession) -> list:
    # Derive a year-month key to use as a secondary partition dimension
    year_month = (
        extract("year", Transaction.created_at) * 100
        + extract("month", Transaction.created_at)
    ).label("year_month")

    stmt = select(
        Transaction.account_id,
        Transaction.created_at,
        Transaction.amount,
        year_month,
        func.sum(Transaction.amount)
        .over(
            # Partition by BOTH account AND calendar month
            partition_by=[
                Transaction.account_id,
                extract("year", Transaction.created_at),
                extract("month", Transaction.created_at),
            ],
            order_by=Transaction.created_at,
            rows=(None, 0),
        )
        .label("monthly_running_total"),
    ).order_by(Transaction.account_id, Transaction.created_at)

    result = await session.execute(stmt)
    return result.mappings().all()

By adding the year and month extracts to partition_by, the window resets at every month boundary without any Python-side logic. The database handles the boundary detection entirely through the partition key — each unique (account_id, year, month) tuple starts a fresh frame at UNBOUNDED PRECEDING. This approach scales to any reset boundary: fiscal quarters, billing cycles, or custom date bins, as long as you can express the boundary as a column or derived expression.

Frequently Asked Questions

from sqlalchemy import func, select
from sqlalchemy.dialects import postgresql

stmt = select(
    Transaction.account_id,
    func.sum(Transaction.amount)
    .over(
        partition_by=Transaction.account_id,
        order_by=Transaction.created_at,
        rows=(None, 0),
    )
    .label("running_total"),
)
# Print PostgreSQL-dialect SQL without executing
print(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))

Related

• Window Functions and Analytical Queries — parent guide covering all window function types, frame clauses, combining with GROUP BY, and mapping results.
• Common Table Expressions (CTEs) and Recursive Queries — chaining running total CTEs into multi-stage analytical pipelines.
• High-Performance Bulk Inserts and Updates — persisting incremental running total results using insert().from_select() with minimal roundtrips.
• Advanced Query Patterns and Bulk Data Operations — top-level guide to SQLAlchemy 2.0 query patterns for high-throughput workloads.