Skip to content

Streaming Large Result Sets with asyncpg Cursors

await conn.fetch("SELECT * FROM events") is the first thing anyone writes with asyncpg, and it is the right call for a query that returns a page of rows. It is the wrong call for an export, a backfill, a nightly reconciliation or anything else whose row count grows with the business, because it decodes the entire result into a Python list before your first line of processing code runs. On the table used throughout this guide — 200,000 rows holding 42.7 MB of payload text — that costs 70.3 MiB of peak Python memory, measured with tracemalloc. The same query through a server-side cursor costs 0.8 MiB, because only one batch exists at a time. This guide shows how to make that switch, how to size the batch, and the four constraints a cursor imposes in return.

Prerequisites

  • Python 3.11+ and asyncpg (pip install asyncpg) with a reachable PostgreSQL server. Every number here came from asyncpg 0.31 against PostgreSQL 18 over a loopback connection.
  • Driver basics from Async Database Drivers, especially how a connection pool hands out connections.
  • Async iteration from writing async iterators for paginated APIs, since a cursor is consumed with async for.
Peak Python memory for the same 200,000 rows 2 bars comparing conn.fetch(query) with the others. Peak Python memory for the same 200,000 rows conn.fetch(query) 70.3 MiB cursor, prefetch=1000 0.8 MiB tracemalloc peak on this machine; 200,000 rows totalling 42.7 MB of payload text. The cursor holds one batch, so its memory does not grow with the result set.

1. Measure what fetching everything costs

Before changing anything, put a number on the problem. tracemalloc reports the peak Python-object memory of a block, which is exactly what the row list occupies:

import asyncio
import tracemalloc

import asyncpg

QUERY = "SELECT id, payload FROM events"


async def fetch_all(dsn: str) -> None:
    conn = await asyncpg.connect(dsn)
    tracemalloc.start()
    rows = await conn.fetch(QUERY)                       # every row decoded up front
    total = sum(len(r["payload"]) for r in rows)
    peak = tracemalloc.get_traced_memory()[1] / 1024 / 1024
    print(f"rows={len(rows)} bytes={total} peak={peak:.1f} MiB")
    await conn.close()

Run against the 200,000-row table this prints rows=200000 bytes=42688890 peak=70.3 MiB. The peak is larger than the payload itself because each row becomes a Record object holding decoded Python strings. The figure scales linearly: ten times the rows is roughly ten times the memory, and a service that does this in a request handler under concurrency multiplies it again by the number of concurrent handlers.

Verify: the reported peak grows in proportion to the row count when you change the LIMIT.

2. Stream the same query through a cursor

A cursor asks PostgreSQL to keep the result on the server and hand it over in batches. In asyncpg, Connection.cursor() returns a factory that is iterated with async for, and it must run inside a transaction:

async def stream_all(dsn: str) -> None:
    conn = await asyncpg.connect(dsn)
    tracemalloc.start()
    rows = 0
    total = 0
    async with conn.transaction():                       # a portal lives inside a transaction
        async for record in conn.cursor(QUERY, prefetch=1000):
            rows += 1
            total += len(record["payload"])
    peak = tracemalloc.get_traced_memory()[1] / 1024 / 1024
    print(f"rows={rows} bytes={total} peak={peak:.1f} MiB")
    await conn.close()

This prints rows=200000 bytes=42688890 peak=0.8 MiB — the same rows, the same bytes, 88 times less peak memory. Wall-clock time went from 0.38 s to 0.53 s, the cost of 200 round trips instead of one, which is a good trade whenever the result set is large enough to matter.

Forget the transaction and asyncpg tells you plainly:

asyncpg.exceptions.NoActiveSQLTransactionError: cursor cannot be created outside of a transaction

Verify: the peak stays flat as you raise the row count, while step 1's peak keeps climbing.

How a row reaches your loop through a cursor 5 stages from BEGIN to your loop. How a row reaches your loop through a cursor BEGIN a portal needs it DECLARE portal query planned once FETCH n rows one round trip per batch decode batch n records in memory your loop next batch on demand The server keeps the portal open for the life of the transaction, not the connection.

3. Size the prefetch for round trips, not memory

prefetch controls how many rows asyncpg requests per FETCH. It is the only tuning knob that matters, and its effect is entirely about round trips. Streaming 50,000 rows over loopback:

prefetch round trips elapsed
1 50,000 3.42 s
10 5,000 0.38 s
100 500 0.07 s
1,000 50 0.04 s
10,000 5 0.04 s

The default of 50 already avoids the pathological case. Below about 100 the per-round-trip latency dominates; above about 1,000 you are buying nothing but a larger batch in memory. On a real network — a managed database a millisecond away rather than a loopback socket — the left-hand rows get dramatically worse, so the practical advice is: leave it alone, or set it to a round number between 100 and 1,000 that matches the batch size your processing code wants anyway.

Verify: time the same query at two prefetch sizes; the ratio should track the round-trip count until it flattens.

Prefetch size against wall-clock time 5 bars comparing prefetch=1 with the others. Prefetch size against wall-clock time prefetch=1 3.42 s prefetch=10 0.38 s prefetch=100 0.07 s prefetch=1000 0.04 s prefetch=10000 0.04 s Same 50,000 rows over a loopback connection; a real network widens the gap at small sizes. Round trips dominate below about 100; past 1,000 you only buy more memory per batch.

4. Process in batches rather than row by row

Most streaming jobs write somewhere else — another table, a file, an HTTP endpoint — and those destinations prefer batches. Cursor.fetch(n) returns a list and an empty list at the end, which makes a clean while loop:

async def export_batches(conn, batch_size: int = 50_000):
    async with conn.transaction():
        cursor = await conn.cursor("SELECT id FROM events ORDER BY id")
        while batch := await cursor.fetch(batch_size):   # empty list ends the stream
            print("batch of", len(batch), "first id", batch[0]["id"])

Against the same table this prints four lines, first ids 1, 50001, 100001 and 150001. The sibling methods are fetchrow() for a single record and forward(n), which skips n rows server-side and returns how many it actually moved — useful for resuming a job at a known offset without transferring the rows you are skipping.

An async generator is the more composable shape, because the caller never sees the cursor at all:

async def export_rows(conn, batch_size: int = 1000):
    """Stream a large table, yielding lists of records."""
    batch = []
    async with conn.transaction():
        async for record in conn.cursor(QUERY, prefetch=batch_size):
            batch.append(record)
            if len(batch) == batch_size:
                yield batch
                batch = []
    if batch:                                            # the final short batch
        yield batch

Consuming it with async for batch in export_rows(conn) streamed all 200,000 rows at a 1.2 MiB peak. Note that the transaction is held open across yield, so the caller must consume the generator to completion or close it explicitly — see closing async generators with aclosing for why an abandoned generator leaves that transaction open until garbage collection.

Verify: the batch sizes sum to the table's row count, and the last batch is the only short one.

5. Consume inside the pool acquire

A cursor is bound to its connection, and a pooled connection goes back to the pool at the end of the async with. Returning a cursor from a helper that acquires and releases produces a failure that is confusing out of context:

asyncpg.exceptions._base.InterfaceError: cannot call CursorFactory.__aiter__():
the underlying connection has been released back to the pool

The fix is structural: the acquire, the transaction and the consumption all live in one block.

async def stream_from_pool(pool) -> int:
    total = 0
    async with pool.acquire() as conn, conn.transaction():
        async for record in conn.cursor(QUERY, prefetch=2000):
            total += len(record["payload"])              # consumed before the release
    return total

That returned 42688890 bytes, matching the non-pooled runs exactly. Remember that the connection is pinned for the whole stream: a long export with max_size=4 leaves three connections for everything else, so give background exports their own small pool rather than letting them compete with request traffic. The reasoning is the same as bulkheads — isolate the slow work so it cannot exhaust the shared resource.

Verify: the byte total from the pooled version equals the direct-connection version.

6. Keep the transaction short enough to survive

The portal lives inside a transaction, and a transaction that sits idle is a liability. Most production servers set idle_in_transaction_session_timeout for exactly this reason. With it set to one second, a consumer that pauses for two seconds between batches loses the connection:

conn = await asyncpg.connect(dsn, server_settings={"idle_in_transaction_session_timeout": "1000"})

first batch: [1, 2]
after idle 2s -> InterfaceError : cannot call Cursor.fetch(): the underlying connection is closed

The server log is explicit: FATAL: terminating connection due to idle-in-transaction timeout. asyncpg surfaces it only as a closed connection, which is why this failure is usually misread as a network problem.

Three ways to stay under the limit, in order of preference:

  1. Do the slow work outside the loop. Collect a batch, close nothing, but push the HTTP call or the file write into a task whose result you await after the transaction ends, or use a bounded queue with a consumer task as in batching queue items by size and time.
  2. Chunk by key instead of by cursor. For a table with a monotonic id, WHERE id > $1 ORDER BY id LIMIT n in a loop gives you restartability and no long transaction, at the cost of re-planning each query.
  3. Raise the timeout for that session only, via server_settings on a dedicated export connection — a deliberate, local exception rather than a server-wide one.

Verify: with a short idle_in_transaction_session_timeout and a sleep in the loop, the stream fails; moving the sleep outside the transaction makes it pass.

What a server-side cursor requires A grid of 4 rows by 2 columns. What a server-side cursor requires constraint what happens if you ignore it the fix an open transaction NoActiveSQLTransactionError async with conn.transaction() the connection is pinned no other query on it acquire a second connection pool release ends it InterfaceError on the next row consume inside the acquire idle transactions are killed connection closed mid-stream do slow work outside the loop Every one of these is a consequence of the portal living inside the transaction.

Verification

A streaming read is correct when:

  • Memory is flat: the tracemalloc peak does not grow when the row count does.
  • Row counts match: the streamed count and byte total equal what conn.fetch() returned on the same query.
  • Round trips are sane: rows divided by prefetch is in the tens or hundreds, not the tens of thousands.
  • Nothing outlives its scope: no cursor is used after its acquire block, and no generator holding a transaction is abandoned.
  • The transaction stays busy: there is no await on an external service between batches inside the transaction.

Pitfalls & edge cases

  • prefetch with a manual fetch(n). Passing prefetch and then calling cursor.fetch(n) mixes two batch sizes; pick one. asyncpg raises InterfaceError if you pass prefetch to a cursor you then drive manually.
  • Sorting without an index. A cursor does not make ORDER BY incremental — PostgreSQL still materialises and sorts the whole set server-side before the first row arrives. The client memory is saved; the server work is not.
  • Cancellation mid-stream. Cancelling the consuming task unwinds through the async with, which rolls back the transaction and drops the portal. That is correct, but any partial work you did elsewhere is not rolled back — see cancellation patterns.
  • Reading and writing on one connection. The cursor pins the connection, so writes based on the streamed rows need a second connection, which puts them in a different transaction. Plan for the reader seeing a snapshot older than the writer's.
  • COPY is faster for a plain dump. If the destination is a file or another table, copy_from_query and copy_records_to_table skip Python object creation entirely.

Frequently Asked Questions

How do I iterate over a large query result in asyncpg without loading it all?

Open a transaction and iterate the cursor factory: async with conn.transaction(): async for record in conn.cursor(query, prefetch=1000). asyncpg fetches prefetch rows per round trip and decodes only that batch, so peak memory stays constant regardless of the result size.

Why does asyncpg raise NoActiveSQLTransactionError when I create a cursor?

A server-side cursor is a portal, and PostgreSQL portals only exist inside a transaction. Wrap the iteration in async with conn.transaction(). asyncpg deliberately does not open one implicitly, because the transaction's lifetime is a decision with consequences for locking and for idle-in-transaction timeouts.

What prefetch value should I use with an asyncpg cursor?

Something between 100 and 1,000 for most workloads. The value only controls how many rows come back per round trip: measured on 50,000 rows, prefetch=1 took 3.42 s, prefetch=100 took 0.07 s, and prefetch=1000 took 0.04 s, after which more prefetch buys nothing but memory per batch.

Can I use an asyncpg cursor with a connection pool?

Yes, as long as the cursor is consumed inside the same async with pool.acquire() block. Once the connection returns to the pool, using the cursor raises InterfaceError. Also remember the connection is pinned for the whole stream, so long exports deserve their own small pool.

Why does my long asyncpg stream fail with 'the underlying connection is closed'?

Usually the server terminated the session for exceeding idle_in_transaction_session_timeout, which the PostgreSQL log records as "terminating connection due to idle-in-transaction timeout". It happens when the consumer does slow work between batches. Move that work outside the transaction, or page by key instead of holding a cursor open.