Async Python with a sync database driver is a common bug. The driver blocks the event loop; everything stalls. Picking the right async driver matters. This post is the working comparison.
Postgres: asyncpg vs psycopg 3
asyncpg
- Raw performance: 2–3× faster than psycopg in micro-benchmarks.
- Native async; not a wrapper.
- Direct API; less flexible than psycopg.
- Works with SQLAlchemy via
asyncpgdialect.
import asyncpg
pool = await asyncpg.create_pool(dsn, min_size=2, max_size=10)
async with pool.acquire() as conn:
user = await conn.fetchrow("SELECT * FROM users WHERE id = $1", 42)
psycopg 3 async
- Fewer micro-benchmark wins; production workloads similar.
- Same API surface as sync psycopg; easier to share code.
- Better feature parity (COPY, server cursors, etc.).
- Works with SQLAlchemy.
import psycopg
from psycopg.rows import dict_row
async with await psycopg.AsyncConnection.connect(dsn) as conn:
async with conn.cursor(row_factory=dict_row) as cur:
await cur.execute("SELECT * FROM users WHERE id = %s", [42])
user = await cur.fetchone()
Pick
- Raw queries, max perf: asyncpg.
- SQLAlchemy or Pydantic-heavy: either; small difference.
- Migration from sync psycopg: psycopg 3 (less code change).
For typical SQLAlchemy 2.0 async usage, both work; asyncpg is the default.
MySQL: asyncmy vs aiomysql
- asyncmy: faster, more active.
- aiomysql: older, well-known.
In 2026, asyncmy is the default for new code. Both work with SQLAlchemy.
SQLite: aiosqlite, libsql
For local SQLite:
import aiosqlite
async with aiosqlite.connect("app.db") as db:
async with db.execute("SELECT * FROM users") as cursor:
rows = await cursor.fetchall()
For Turso / libsql:
import libsql_experimental as libsql
conn = libsql.connect("local.db", sync_url=TURSO_URL, auth_token=TOKEN)
conn.sync()
rows = conn.execute("SELECT * FROM users").fetchall()
Note: libsql_experimental is sync. For async, use the libsql HTTP client or run sync in a threadpool.
For SQLite at the Edge in 2026 the broader picture.
Connection pooling
Always pool. Cold connections are expensive (TLS handshake, auth):
# asyncpg
pool = await asyncpg.create_pool(
dsn,
min_size=2, max_size=10,
command_timeout=10,
server_settings={"jit": "off"}, # disable JIT for short queries
)
Pool sizes:
- Bound by Postgres
max_connections: set per-app pool low (10–25), use external pooler (PgBouncer) for many apps. - Per-process pool: matters with multi-worker setups.
Statement caching
asyncpg caches prepared statements per connection. For dynamic SQL with many shapes, the cache fills:
pool = await asyncpg.create_pool(dsn, statement_cache_size=200)
For high-cardinality query shapes, set the cache appropriately.
Pipelining (Postgres 14+)
Send multiple queries in a single round-trip:
# asyncpg
async with pool.acquire() as conn:
async with conn.transaction():
await asyncio.gather(
conn.execute("INSERT INTO ..."),
conn.execute("INSERT INTO ..."),
conn.execute("INSERT INTO ..."),
)
asyncpg pipelines automatically. Big win for bulk-write workloads.
Common mistakes
1. Sync driver in async code
psycopg2 calls block the event loop. Use psycopg 3 async or asyncpg. Verify there’s no psycopg2 sneaking in via dependencies.
2. Per-request connections
Without a pool, you pay TLS handshake on every request. Always pool.
3. Ignoring command_timeout
A slow query holds the connection forever. Set a timeout.
4. Not awaiting
result = pool.fetchrow(...) # ⛔ coroutine not awaited
Linters catch this; debug mode warns.
5. Mixing sync and async sessions in SQLAlchemy
Async session needs async engine + async driver. Mixing produces silent bugs.
Read this next
- SQLAlchemy 2.0 Deep Patterns
- Modern AsyncIO Patterns
- FastAPI + Pydantic v2 + SQLAlchemy 2.0
- PostgreSQL Performance Tuning
If you want my asyncpg + SQLAlchemy 2.0 + connection pool reference, it’s at rajpoot.dev .
Building something AI-, backend-, or data-heavy and want a second pair of eyes? I do consulting and freelance work — see my projects and ways to reach me at rajpoot.dev .