asyncpg
asyncpg is a PostgreSQL driver for asyncio that speaks the Postgres binary wire protocol directly instead of wrapping libpq. Skipping libpq and the DB-API layer is the whole design: it decodes results straight into Python objects, caches prepared statements per connection, and exposes Postgres features such as scrollable cursors, COPY, and LISTEN/NOTIFY as first-class methods rather than as escape hatches. You get a small API: connect() or create_pool(), then execute, fetch, fetchrow, and fetchval, with numbered $1 placeholders. Rows come back as Record objects that behave like both a tuple and a read-only mapping. It requires Python 3.9 or later and supports PostgreSQL 9.5 through 18. The README reports asyncpg averaging 5x faster than psycopg3 in the maintainers' own benchmark toolkit run in June 2023.
For an asyncio service talking to Postgres and nothing else, asyncpg is the fastest and cleanest option and has been for years. Check your connection topology first: if PgBouncer sits in the middle in transaction pooling mode, psycopg 3 will give you a less irritating time for a small cost in throughput.
Use it if
- Your service is already asyncio end to end (FastAPI, aiohttp, Litestar) and a synchronous driver would block the event loop on every query
- You are pushing serious query volume and the per-query overhead of libpq plus DB-API row conversion is showing up in profiles
- You want Postgres-specific features without ceremony: COPY in and out via copy_records_to_table, server-side cursors for streaming millions of rows, LISTEN/NOTIFY through add_listener
- You use rich Postgres types. Arrays, composite types, ranges, hstore, and combinations of those decode automatically, and set_type_codec lets you map a custom type to a Python class in one call
- You write SQL by hand and do not want an ORM in the way; the API is thin enough that the query is the only thing you maintain
- Your codebase is synchronous. asyncpg is asyncio-only, and sprinkling asyncio.run around a WSGI app is worse than just using psycopg 3, which offers both sync and async from one package
- You sit behind PgBouncer in transaction or statement pooling mode. Prepared statement caching, which is where much of the speed comes from, breaks there, and the documented fix is statement_cache_size=0, which gives most of that advantage back
- You need DB-API 2.0 compatibility. asyncpg deliberately does not implement it, so tooling that expects a standard cursor, and any sync SQLAlchemy or Django setup, will not work against it
- You want a driver that coerces types for you. Passing a str for an integer column raises DataError rather than casting, and json and jsonb come back as raw strings until you register a codec, which surprises people migrating from psycopg
- You need portability across databases. This is Postgres and only Postgres; other servers that speak the protocol may work but are explicitly not tested
- You want a predictable release train. Recent history is roughly one release a year (0.29.0 in November 2023, 0.30.0 in October 2024, 0.31.0 in November 2025), the project is still on 0.x so minor bumps carry breaking changes, and there are 252 open issues (290 counting PRs)
Setup reality
pip install asyncpg has no dependencies at all on Python 3.11 and up (3.9 and 3.10 pull async_timeout), and prebuilt wheels cover CPython 3.10 through 3.14 on manylinux and musllinux x86_64 and aarch64, macOS x86_64 and arm64, and Windows. On anything outside that list, including PyPy and unusual libc combinations, pip falls back to compiling the Cython extension from source, which needs a C compiler and Python development headers. GSSAPI or SSPI authentication is an extra: pip install 'asyncpg[gssauth]', which drags in gssapi on Unix and needs Kerberos development headers there. Two things bite in the first hour. Placeholders are numbered $1, $2, not %s and not named, so every query copied from a psycopg codebase has to be rewritten. And connection pool objects are async context managers whose sizes default to min_size=10 and max_size=10, which means ten connections per worker process; multiply that by your worker count before Postgres starts rejecting connections.
Patterns
Open a connection and run queriesconnect-and-query
import asyncio
import asyncpg
async def main():
conn = await asyncpg.connect('postgresql://user:pass@localhost/db')
try:
rows = await conn.fetch('SELECT id, email FROM users WHERE age > $1', 21)
one = await conn.fetchrow('SELECT * FROM users WHERE id = $1', 1)
count = await conn.fetchval('SELECT count(*) FROM users')
await conn.execute('UPDATE users SET seen = now() WHERE id = $1', 1)
finally:
await conn.close()
asyncio.run(main())Placeholders are $1, $2 in position order, never %s and never named. fetchrow returns None when nothing matches and fetchval returns the first column of the first row, so both need a None check before you index into them.
Use a pool instead of one connection per requestconnection-pool
pool = await asyncpg.create_pool(
dsn='postgresql://user:pass@localhost/db',
min_size=2,
max_size=10,
command_timeout=30,
)
async with pool.acquire() as conn:
rows = await conn.fetch('SELECT 1')
# short queries can skip acquire entirely
rows = await pool.fetch('SELECT 1')
await pool.close()min_size and max_size both default to 10, so the pool opens ten connections immediately per process. With several uvicorn workers that multiplies fast and hits max_connections on the server. Set both explicitly.
Work with the Record type rows come back asread-record-objects
row = await conn.fetchrow('SELECT id, email FROM users WHERE id = $1', 1)
row['email'] # by column name
row[1] # by position
dict(row) # {'id': 1, 'email': '...'}
list(row.keys()) # ['id', 'email']
if row is None:
raise LookupError('no such user')Record is immutable and is not a dict, so json.dumps(row) fails and row.get('x') does not exist. Convert with dict(row) before serializing, and remember that duplicate column names in a join collapse when you do.
Wrap statements in a transactiontransactions
async with pool.acquire() as conn:
async with conn.transaction():
await conn.execute('INSERT INTO accounts(id, bal) VALUES ($1, $2)', 1, 100)
await conn.execute('UPDATE ledger SET total = total + $1', 100)
# explicit isolation and read-only
async with conn.transaction(isolation='serializable', readonly=True):
...Leaving the block normally commits, and any exception rolls back. Nested transaction() blocks become savepoints rather than errors, which is convenient but means an inner rollback does not undo the outer block.
Load many rows with COPY instead of INSERTbulk-insert-copy
records = [(1, 'a@example.com'), (2, 'b@example.com')]
await conn.copy_records_to_table(
'users',
records=records,
columns=['id', 'email'],
)
# or straight from a CSV file
await conn.copy_to_table('users', source='users.csv', format='csv', header=True)COPY is far faster than executemany for large batches, but it bypasses ON CONFLICT entirely, so a duplicate key aborts the whole load. For upserts, COPY into a temporary table and then INSERT ... SELECT ... ON CONFLICT from it.
Run the same statement over many parameter setsbatch-executemany
await conn.executemany(
'INSERT INTO events(user_id, kind) VALUES ($1, $2)',
[(1, 'login'), (2, 'logout'), (3, 'login')],
)executemany returns nothing, so you cannot use RETURNING with it, and it runs atomically: one bad row rolls the whole batch back. For tens of thousands of rows copy_records_to_table is the faster tool.
Iterate a huge result set without loading it allstream-large-results
async with conn.transaction():
async for row in conn.cursor('SELECT * FROM big_table', prefetch=1000):
handle(row)
# or take a page at a time
async with conn.transaction():
cur = await conn.cursor('SELECT * FROM big_table')
batch = await cur.fetch(500)Server-side cursors only work inside an explicit transaction; calling cursor() outside one raises InterfaceError. fetch() on a connection buffers every row in memory, which is what you are avoiding here.
Get dicts back from json and jsonb columnsjson-type-codec
import json
async def init_conn(conn):
for typename in ('json', 'jsonb'):
await conn.set_type_codec(
typename,
encoder=json.dumps,
decoder=json.loads,
schema='pg_catalog',
)
pool = await asyncpg.create_pool(dsn, init=init_conn)By default json and jsonb decode to str, so row['payload']['id'] raises TypeError until you register this. Passing init to create_pool applies it to every connection the pool opens, including replacements for dropped ones.
Catch specific Postgres errorshandle-postgres-errors
import asyncpg
try:
await conn.execute('INSERT INTO users(email) VALUES ($1)', email)
except asyncpg.UniqueViolationError as exc:
raise Conflict(f'{exc.constraint_name} already taken') from exc
except asyncpg.ForeignKeyViolationError:
raise BadRequest('referenced row does not exist')
except asyncpg.PostgresError as exc:
log.exception('sqlstate=%s', exc.sqlstate)asyncpg generates one exception class per SQLSTATE, all under PostgresError, and each carries detail, hint, constraint_name, and table_name. That is much more precise than string-matching an error message, which is what DB-API drivers push you toward.
Subscribe to Postgres notificationslisten-notify
def on_event(connection, pid, channel, payload):
print(channel, payload)
conn = await asyncpg.connect(dsn)
await conn.add_listener('jobs', on_event)
# elsewhere: SELECT pg_notify('jobs', '{"id": 7}');
await conn.remove_listener('jobs', on_event)The callback is synchronous, so anything slow in it stalls the connection's read loop; schedule real work with asyncio.create_task. Listener connections must be dedicated and long-lived, not taken from a pool that recycles them.
Connect through PgBouncer in transaction pooling modepgbouncer-mode
pool = await asyncpg.create_pool(
dsn='postgresql://user:pass@pgbouncer:6432/db',
statement_cache_size=0,
max_cacheable_statement_size=0,
min_size=2,
max_size=10,
)Without statement_cache_size=0 you get intermittent 'prepared statement already exists' errors, because PgBouncer hands the same session to different clients. Turning the cache off also removes most of asyncpg's performance edge, so weigh psycopg 3 at that point.
Return your own row objectscustom-record-class
class DictRecord(asyncpg.Record):
def __getattr__(self, name):
try:
return self[name]
except KeyError as exc:
raise AttributeError(name) from exc
pool = await asyncpg.create_pool(dsn, record_class=DictRecord)
row = await pool.fetchrow('SELECT id, email FROM users LIMIT 1')
row.emailrecord_class must subclass asyncpg.Record; a plain dataclass or NamedTuple is rejected. Because Record is a C type you cannot add slots or mutable state, so keep the subclass to accessor sugar.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| psycopg | PyPI | You need sync and async from the same driver, DB-API compatibility, or a working setup behind PgBouncer without disabling statement caching |
| sqlalchemy | PyPI | You want an ORM or query builder on top; its async engine can drive asyncpg while giving you models and migrations |
| databases | PyPI | You want one async query API that can point at Postgres, MySQL, or SQLite without rewriting the data layer |