psycopg3 vs asyncpg for drift pipeline catalog queries

This page compares psycopg 3 and asyncpg for the specific workload of a drift pipeline: reading privilege catalogs from many PostgreSQL hosts, repeatedly, on a schedule — and shows which driver’s connection model, pooling, and cursor behavior fits which shape of that job.

A drift pipeline is I/O-bound fan-out: dozens or hundreds of hosts, each answering a handful of read-only catalog queries against pg_authid, pg_auth_members, and information_schema.role_table_grants. The bottleneck is never CPU; it is how efficiently you multiplex connections and stream rows. psycopg 3 and asyncpg both do this well, but they make opposite ergonomic bets — one gives you a familiar DBAPI surface with an async option bolted on cleanly, the other is async-native and squeezes out more throughput at the cost of DBAPI compatibility. The choice belongs alongside the other decisions in system catalog query optimization.

When to use this — and when not to

Reach for this comparison when:

• You are building or refactoring the extraction stage of a drift pipeline and have to commit to one PostgreSQL driver for the whole fleet.
• Your run time is dominated by round-trips across many hosts and you want to overlap them with concurrency rather than more workers.
• You need server-side cursors to stream large grant sets without materializing them in the client.

Skip this decision when:

• You issue a few queries against one database in a synchronous CLI. Plain psycopg 3 in sync mode is the obviously correct, boring choice.
• Your codebase is already committed to an async framework and ORM that dictates the driver — honor that, don’t fight it.
• You target multiple engines through one abstraction (MySQL, Oracle); a Postgres-only driver comparison is the wrong altitude. Handle that in the cross-DB parser adapters layer instead.

Step-by-step implementation

Step 1 — psycopg 3: pooled catalog reads (sync and async)

psycopg 3 (import psycopg) is DBAPI 2.0-shaped. Connect synchronously with psycopg.connect(), and pool with ConnectionPool from the separate psycopg_pool package. A dict_row factory keeps downstream parsing readable.

from psycopg_pool import ConnectionPool
from psycopg.rows import dict_row

GRANTS_SQL = """
SELECT grantee, privilege_type,
       table_schema || '.' || table_name AS object_scope
FROM information_schema.role_table_grants
WHERE grantee NOT IN ('postgres', 'PUBLIC')
"""

pool = ConnectionPool("host=db1 dbname=app user=drift_ro", min_size=2, max_size=8)

def read_grants(host_conninfo: str) -> list[dict]:
    with pool.connection() as conn:                 # borrowed from the pool
        with conn.cursor(row_factory=dict_row) as cur:
            cur.execute(GRANTS_SQL)
            return cur.fetchall()

The async path mirrors it exactly, which is psycopg 3’s selling point — one mental model, two runtimes. Use AsyncConnectionPool and await:

import asyncio
from psycopg_pool import AsyncConnectionPool
from psycopg.rows import dict_row

async def read_grants_async(pool: AsyncConnectionPool) -> list[dict]:
    async with pool.connection() as conn:
        async with conn.cursor(row_factory=dict_row) as cur:
            await cur.execute(GRANTS_SQL)
            return await cur.fetchall()

Verification — a borrowed connection is returned to the pool on block exit:

print(pool.get_stats()["pool_size"])   # e.g. 2  (min_size), grows on demand

Step 2 — asyncpg: async-native pool and prepared fetch

asyncpg has no sync mode and is not DBAPI. Connect with await asyncpg.connect(), pool with await asyncpg.create_pool(), and read with conn.fetch(), which prepares and caches the statement automatically. Parameters are positional $1, $2 — not the %s psycopg uses.

import asyncio
import asyncpg

async def build_pool(dsn: str) -> asyncpg.Pool:
    return await asyncpg.create_pool(dsn, min_size=2, max_size=8)

async def read_grants(pool: asyncpg.Pool) -> list[asyncpg.Record]:
    async with pool.acquire() as conn:              # borrowed connection
        return await conn.fetch(
            """
            SELECT grantee, privilege_type,
                   table_schema || '.' || table_name AS object_scope
            FROM information_schema.role_table_grants
            WHERE grantee <> $1
            """,
            "postgres",
        )

Verification — Record rows support both index and key access:

async def main():
    pool = await build_pool("postgres://drift_ro@db1/app")
    rows = await read_grants(pool)
    print(rows[0]["grantee"], rows[0][1])   # key access and positional access
    await pool.close()

Step 3 — Stream large grant sets with a server-side cursor

For a host with tens of thousands of grants, do not materialize the whole result. psycopg 3 opens a server-side (named) cursor and batches with itersize:

def stream_grants(conn, batch=2000):
    with conn.cursor(name="grants_srv") as cur:     # named ⇒ server-side
        cur.itersize = batch
        cur.execute(GRANTS_SQL)
        for row in cur:                             # streamed, not buffered
            yield row

asyncpg’s server-side cursor must live inside a transaction and is iterated with async for:

async def stream_grants(conn):
    async with conn.transaction():                  # required for a cursor
        async for row in conn.cursor(GRANTS_SQL):
            yield row

Verification — peak client memory stays flat regardless of row count, because neither path buffers the full result set client-side.

Worked example: 120 hosts, hourly catalog sweep

Scenario: 120 PostgreSQL 15 hosts, a read-only drift_ro role, one grant query and one role-membership query per host, run hourly. Each query round-trips ~15 ms. Run serially that is 120 × 2 × 15 ms ≈ 3.6 s of pure latency, ignoring connect cost. With an async pool bounded to 20 concurrent connections, the round-trips overlap and wall-clock collapses toward the slowest host plus scheduling overhead.

import asyncio, asyncpg

async def sweep(dsns: list[str], concurrency: int = 20):
    sem = asyncio.Semaphore(concurrency)
    async def one(dsn):
        async with sem:
            conn = await asyncpg.connect(dsn)
            try:
                return dsn, await conn.fetch(
                    "SELECT grantee, privilege_type "
                    "FROM information_schema.role_table_grants "
                    "WHERE grantee <> $1", "postgres")
            finally:
                await conn.close()
    return dict(await asyncio.gather(*(one(d) for d in dsns)))

Expected shape of the result:

{'postgres://drift_ro@host001/app': [, ...],
 'postgres://drift_ro@host002/app': [, ...],
 ...}

The same sweep in psycopg 3 with AsyncConnectionPool reads almost identically; asyncpg edges ahead on raw throughput because of its binary protocol and lighter row objects, while psycopg 3 wins if any part of this pipeline still runs synchronously and you want one driver for both. For batching mechanics beyond a single sweep, see async privilege batching.

Gotchas and engine-specific notes

Parameter style differs and is not interchangeable. psycopg 3 uses %s placeholders; asyncpg uses $1. Copy a query between drivers without translating placeholders and it fails at execute time. Keep engine-specific SQL constants separate.

Server-side cursors need a live transaction in asyncpg. conn.cursor() outside a transaction raises. psycopg 3’s named cursor manages its own transaction. Wrap asyncpg streaming in async with conn.transaction(): or nothing streams.

asyncpg is PostgreSQL-only and not DBAPI. No dict rows by default, no %s, no sync API, and it will never talk to MySQL. If your fleet is heterogeneous, the driver-neutral logic must live above the adapter boundary, not in the query code.

Pool sizing throttles the server, not just the client. 200 concurrent connections against a small instance will exhaust max_connections and page the DBA. Bound the pool (max_size) and add a semaphore for fan-out; a read-only sweep rarely needs more than 10–20 live connections per instance.

PUBLIC and bootstrap roles pass through both drivers unchanged. Neither driver filters catalog noise — exclude PUBLIC and postgres in SQL, or every host reports the same built-in grants.

Compliance note

Driver choice is invisible to an auditor, but it underwrites a control they do care about: that the extraction identity is strictly read-only. Both psycopg 3 and asyncpg should connect as a dedicated drift_ro role holding only SELECT on the catalog views, satisfying least-privilege collection under PCI-DSS Requirement 7 and SOC 2 CC6.1. Recording the driver, pool bounds, and the read-only connection role in the run metadata gives reviewers a reproducible, non-mutating collection method — the pipeline can never alter the very grants it inspects, which is the property that makes its output admissible as evidence for the drift detection engines baseline.

Frequently asked questions

Which driver is faster for catalog reads? asyncpg generally posts higher raw throughput because of its binary protocol and lightweight Record rows. For a latency-bound fan-out across many hosts, the difference is usually dwarfed by network round-trips, so pick on ergonomics unless profiling says otherwise.

Can I use psycopg 3 without async? Yes. psycopg.connect() and ConnectionPool from psycopg_pool are fully synchronous. psycopg 3’s advantage is that the async API mirrors the sync one, so you can start sync and add asyncio later without switching drivers.

How do I stream a huge grant set without exhausting memory? Use a server-side cursor. In psycopg 3, open a named cursor and set itersize. In asyncpg, iterate conn.cursor inside a transaction with async for. Both keep client memory flat by fetching in batches.

Does asyncpg work with MySQL or Oracle? No. asyncpg speaks the PostgreSQL wire protocol only. For a mixed fleet, keep the driver behind a parser-adapter boundary so the diff logic never depends on a Postgres-only client.

Related

• System Catalog Query Optimization — the tuning context this decision sits in
• Python Scripts for Async Batch Privilege Scraping — fan-out batching patterns for either driver
• Async Privilege Batching — bounding concurrency across a fleet

Up: System Catalog Query Optimization