Skip to content

Using psycopg 3 Async Connections and Pools

psycopg 3 is the PostgreSQL driver most Python code already uses, and its async API is the same API with await in front: AsyncConnection, AsyncCursor, and AsyncConnectionPool from the separate psycopg_pool package. That familiarity hides two behaviours worth knowing before you put it under load. A single connection executes one query at a time — measured against PostgreSQL 17, 100 concurrent 10 ms queries on one AsyncConnection took 1.04 s, exactly serial, while a pool of ten connections ran them in 0.12 s. And round trips dominate small statements — 2,000 single-row inserts took 1.93 s one by one, 0.12 s in pipeline mode, and 0.09 s with executemany. This guide sets up connections and pools properly, uses pipelines, and handles pool exhaustion.

Prerequisites

  • Python 3.11+, pip install "psycopg[binary]" psycopg-pool; measured with psycopg 3.3 and psycopg-pool 3.3 against PostgreSQL 17.
  • Pool sizing, from sizing async connection pools for throughput.
  • On Windows, psycopg's async mode needs the selector event loop, not the default proactor loop.

1. Open an async connection

AsyncConnection.connect is a coroutine; the connection is an async context manager that commits on success and rolls back on error when used with async with:

import psycopg


async def count_orders(dsn: str) -> int:
    async with await psycopg.AsyncConnection.connect(dsn) as conn:
        async with conn.cursor() as cur:
            await cur.execute("select count(*) from orders where status = %s", ("open",))
            (n,) = await cur.fetchone()
            return n

Note the async with await: connect() returns a coroutine that must be awaited to get the connection, which is then used as a context manager. Parameters are always passed separately with %s placeholders — never formatted into the SQL string. By default psycopg opens a transaction on the first statement; pass autocommit=True for connections that run independent statements, such as a pool used by read-only handlers.

Verify: a query with a parameter runs, and leaving the block closes the connection (pg_stat_activity no longer shows it).

2. Do not share one connection between concurrent tasks

A PostgreSQL connection runs one statement at a time. psycopg serializes concurrent use of the same AsyncConnection with an internal lock, so it is safe, but it is not concurrent:

async with await psycopg.AsyncConnection.connect(dsn, autocommit=True) as conn:

    async def query():
        async with conn.cursor() as cur:
            await cur.execute("select pg_sleep(0.01)")

    await asyncio.gather(*(query() for _ in range(100)))   # measured: 1.04 s, one after another

Measured: 100 concurrent 10 ms queries on one connection took 1.04 s — 100 × 10 ms plus round trips. The asyncio.gather gave no speed-up because the database connection, not Python, was the bottleneck. Worse, concurrent tasks sharing a non-autocommit connection share its transaction: one task's error rolls back another's work. Give concurrent tasks separate connections, from a pool.

Verify: time a batch of concurrent queries on one connection; it equals the sum of the query times.

100 concurrent 10 ms queries 2 horizontal bars comparing 1 AsyncConnection with the others. 100 concurrent 10 ms queries 1 AsyncConnection 1.044 s pool of 10 0.121 s psycopg 3.3, psycopg-pool 3.3, PostgreSQL 17 on localhost; select pg_sleep(0.01). One connection runs one statement at a time, however many tasks await it.

3. Use AsyncConnectionPool with explicit open

AsyncConnectionPool keeps connections open and hands one to each async with pool.connection() block, returning it at the end. Create it with open=False and open it in your application's startup, so construction does no I/O:

from psycopg_pool import AsyncConnectionPool

pool = AsyncConnectionPool(
    dsn,
    min_size=4,
    max_size=10,
    timeout=5.0,               # max wait for a connection before PoolTimeout
    max_idle=300,              # close idle extras after 5 min
    max_lifetime=3600,         # recycle connections hourly
    kwargs={"autocommit": True},
    open=False,
)


async def lifespan(app):
    await pool.open(wait=True)          # fail startup if the DB is unreachable
    yield
    await pool.close()


async def get_user(user_id: int):
    async with pool.connection() as conn:
        cur = await conn.execute("select * from users where id = %s", (user_id,))
        return await cur.fetchone()

Measured: the same 100 concurrent queries through a 10-connection pool took 0.12 s. pool.get_stats() reported 90 of the 100 requests queued for a connection — the pool, not the database, was the limit, which is what you want. The open=False plus explicit open pattern matters because opening in the constructor is deprecated in psycopg-pool and would happen at import time. Lifespan wiring is covered in managing startup and shutdown with ASGI lifespan.

Verify: under load, pool.get_stats() shows pool_size at most max_size and requests_waiting returning to zero.

4. Cut round trips with pipeline mode and executemany

Each statement is a network round trip. For many small statements, that round trip is most of the cost. Pipeline mode sends statements without waiting for each result; executemany uses it internally:

async with pool.connection() as conn:
    async with conn.pipeline():
        for row in rows:
            await conn.execute("insert into events values (%s, %s)", row)
    # all statements sent; results synced when the block exits


async with pool.connection() as conn, conn.cursor() as cur:
    await cur.executemany("insert into events values (%s, %s)", rows)

Measured on localhost with 2,000 single-row inserts: 1.93 s one at a time, 0.12 s in a pipeline, 0.09 s with executemany — a 16–21× difference, and larger on a real network where each round trip is longer. For bulk loads of tens of thousands of rows or more, COPY is faster still, as in bulk loading rows with asyncpg copy; psycopg supports it too with cur.copy().

Verify: compare insert time for a batch with and without the pipeline; the pipeline version is several times faster.

2,000 single-row inserts, by method 3 horizontal bars comparing one execute per row with the others. 2,000 single-row inserts, by method one execute per row 1.929 s pipeline mode 0.119 s executemany 0.088 s Localhost PostgreSQL 17, autocommit connection; a remote database widens the gap. Round trips, not SQL, are the cost of many small statements.

5. Handle pool exhaustion and broken connections

When every connection is in use and timeout passes, pool.connection() raises PoolTimeout. That is the signal to shed load, not to retry immediately:

from psycopg_pool import PoolTimeout


async def handler(request):
    try:
        async with pool.connection() as conn:
            ...
    except PoolTimeout:
        return JSONResponse({"error": "busy"}, status_code=503, headers={"Retry-After": "1"})

Measured with a two-connection pool held by long queries and timeout=0.5, the third request raised PoolTimeout after 0.501 s. For broken connections — a database restart, a network blip — pass check=AsyncConnectionPool.check_connection to test each connection before handing it out, at the cost of a round trip, or rely on max_lifetime and reconnect handling, as in reconnecting async database pools after failover. Watch for session state leaking between users: a connection returned to the pool keeps its settings and session-level locks.

Verify: saturate the pool in a test; requests beyond capacity fail fast with 503 after timeout, and the service recovers once load drops.

Which psycopg 3 tool for this access pattern? A decision on What does the code do with 4 outcomes. Which psycopg 3 tool for this access pattern? What does the code do? serves concurrent requests AsyncConnectionPool connection per block long job, LISTEN/NOTIFY dedicated connection outside the pool many small statements pipeline / executemany 16-21x fewer waits loads 10k+ rows COPY cur.copy() Match the connection pattern to the work, not the other way round.

Verification

psycopg 3 is used well when:

  • Concurrent tasks use separate connections from a pool, never one shared connection.
  • The pool opens explicitly at startup and closes at shutdown.
  • Many small statements use pipeline mode or executemany.
  • Pool exhaustion fails fast with a clear error, and its stats are exported.

Diagnostic Hook: export pool.get_stats() periodically — especially requests_waiting, requests_wait_ms and pool_available. Waiting requests while the database's own CPU is low means the pool is too small or connections are held too long; waiting requests while the database is saturated means a bigger pool would make things worse.

Pitfalls & edge cases

  • Sharing one AsyncConnection across tasks. Safe but serial, and a shared transaction.
  • Opening the pool in its constructor. Deprecated; open it in startup.
  • Statement-per-row loops. Use pipelines, executemany or COPY.
  • Proactor event loop on Windows. psycopg async needs the selector loop.

Frequently Asked Questions

Can several asyncio tasks share one psycopg AsyncConnection?

They can without crashing, because psycopg serializes access, but they run one at a time and share one transaction. In testing, 100 concurrent 10 ms queries on one connection took 1.04 s, versus 0.12 s through a pool of ten.

How do I use AsyncConnectionPool in FastAPI?

Create it with open=False, call await pool.open() in the lifespan startup and await pool.close() at shutdown, and use async with pool.connection() in handlers or a yield dependency.

How do I speed up many inserts with psycopg 3?

Use pipeline mode or executemany so statements are not sent one round trip at a time. In testing, 2,000 inserts fell from 1.93 s to 0.12 s with a pipeline and 0.09 s with executemany.

What happens when the psycopg pool runs out of connections?

pool.connection() waits up to the pool's timeout and then raises PoolTimeout. Treat that as overload and return 503 rather than retrying immediately.