Running Transactions Safely with asyncpg Pools¶
A transaction is a claim on a connection, and in an async service that claim is easy to hold far longer than intended. The pattern that causes it is ordinary-looking: open a transaction, call an external API to enrich a row, then write the result. For the duration of that HTTP call the connection is checked out, the transaction is open, and every row it has touched stays locked. Twenty concurrent requests doing that against a twenty-connection pool is a full stop, and the database's own metrics show almost nothing — no slow queries, just idle-in-transaction sessions.
Getting this right is mostly about scope: acquire late, release early, keep anything that is not a database operation outside the transaction, and give every statement a server-side timeout so a blocked write cannot hold locks indefinitely. This guide covers that, plus savepoints for partial rollback, what cancellation does to an in-flight transaction, and retrying serialization failures under REPEATABLE READ.
Prerequisites¶
- Python 3.11+ with asyncpg and PostgreSQL 13+:
pip install asyncpg
- The async database drivers overview, and avoiding event loop blocking with asyncpg for the driver's execution model.
1. Keep the transaction as small as the work¶
Acquire the connection immediately before the statements that need it, and release it immediately after.
import asyncpg
# WRONG: an HTTP call inside the transaction holds a connection and its locks
async def enrich_wrong(pool: asyncpg.Pool, order_id: int) -> None:
async with pool.acquire() as conn:
async with conn.transaction():
row = await conn.fetchrow("SELECT * FROM orders WHERE id = $1", order_id)
data = await http_client.get(f"/enrich/{row['sku']}") # 400 ms held
await conn.execute("UPDATE orders SET meta = $1 WHERE id = $2",
data, order_id)
# RIGHT: read, release, call out, then re-acquire for the write
async def enrich_right(pool: asyncpg.Pool, order_id: int) -> None:
async with pool.acquire() as conn:
row = await conn.fetchrow("SELECT sku FROM orders WHERE id = $1", order_id)
data = await http_client.get(f"/enrich/{row['sku']}") # no connection held
async with pool.acquire() as conn:
async with conn.transaction():
await conn.execute("UPDATE orders SET meta = $1 WHERE id = $2",
data, order_id)
Splitting it costs one extra acquisition (microseconds from a warm pool) and removes a 400 ms hold on a scarce resource. If the read and write must be atomic, take the optimistic route instead: re-check a version column in the UPDATE's WHERE clause and retry on zero rows affected. Verify: query pg_stat_activity for state = 'idle in transaction' under load; it should be empty.
2. Let the context manager own commit and rollback¶
conn.transaction() commits on clean exit and rolls back on any exception, including cancellation.
async def transfer(pool: asyncpg.Pool, src: int, dst: int, amount: int) -> None:
async with pool.acquire() as conn:
async with conn.transaction(): # BEGIN
await conn.execute("UPDATE accounts SET balance = balance - $1 "
"WHERE id = $2", amount, src)
await conn.execute("UPDATE accounts SET balance = balance + $1 "
"WHERE id = $2", amount, dst)
# COMMIT on clean exit; ROLLBACK if anything raised
Manual BEGIN/COMMIT via execute() is the anti-pattern here: an exception between them leaves the connection in an open transaction, and because the pool recycles that connection, the next request inherits it. The context manager cannot leave that state behind. Verify: raise deliberately inside the block, then check the same pooled connection serves a subsequent request cleanly.
3. Use savepoints for partial rollback¶
Nested transaction() blocks become savepoints, so one failing item does not discard the batch.
import asyncpg
async def import_batch(pool: asyncpg.Pool, rows: list[dict]) -> tuple[int, int]:
ok = bad = 0
async with pool.acquire() as conn:
async with conn.transaction(): # outer: all or nothing
for row in rows:
try:
async with conn.transaction(): # SAVEPOINT per row
await conn.execute(
"INSERT INTO items (sku, qty) VALUES ($1, $2)",
row["sku"], row["qty"])
ok += 1
except asyncpg.PostgresError: # ROLLBACK TO SAVEPOINT
bad += 1
return ok, bad
Without the inner block, the first constraint violation aborts the whole transaction and every subsequent statement fails with current transaction is aborted. Savepoints are not free — each is a round trip — so for large batches prefer validating up front or using ON CONFLICT DO NOTHING. Verify: import a batch containing one duplicate key; the good rows commit and the counters match.
4. Bound every statement server-side¶
A client-side timeout stops you waiting; only a server-side one stops the database working.
import asyncpg
async def make_pool() -> asyncpg.Pool:
return await asyncpg.create_pool(
dsn=DSN, min_size=5, max_size=20, command_timeout=5.0,
server_settings={
"statement_timeout": "5000", # ms: server cancels the query
"idle_in_transaction_session_timeout": "10000", # ms: kills held locks
"lock_timeout": "2000", # ms: do not queue on a lock
"application_name": "orders-api", # visible in pg_stat_activity
},
)
idle_in_transaction_session_timeout is the safety net for the failure in step 1: if a transaction is left open because a task was cancelled at the wrong moment, PostgreSQL terminates the session and releases its locks rather than letting them block writers indefinitely. lock_timeout prevents a queue forming behind a long-running lock holder. Verify: run SELECT pg_sleep(10) and confirm the server cancels it at 5 s, not the client at 5 s with the query still running.
5. Retry serialization failures, not everything¶
Under REPEATABLE READ or SERIALIZABLE, conflicts are expected and retryable — but only whole transactions can be retried.
import asyncio
import random
import asyncpg
RETRYABLE = ("40001", "40P01") # serialization_failure, deadlock_detected
async def with_retry(pool: asyncpg.Pool, work, attempts: int = 3):
for attempt in range(attempts):
try:
async with pool.acquire() as conn:
async with conn.transaction(isolation="repeatable_read"):
return await work(conn)
except asyncpg.PostgresError as exc:
if getattr(exc, "sqlstate", None) not in RETRYABLE or attempt == attempts - 1:
raise
await asyncio.sleep((2 ** attempt) * 0.05 * random.uniform(0.5, 1.5))
raise RuntimeError("unreachable")
Retry the whole transaction, never individual statements: once a transaction is aborted, every statement in it must run again against the new snapshot. Jittered backoff matters because two conflicting transactions retrying in lockstep will conflict again. Restrict retries to 40001 and 40P01; a unique-violation is a business error and retrying it just fails twice. Verify: run two concurrent conflicting transactions and confirm one retries and succeeds while the error counter records exactly one conflict.
Verification¶
- No idle-in-transaction sessions.
pg_stat_activityshows none under sustained load. - Locks are bounded. No lock wait exceeds
lock_timeout, and no statement outlivesstatement_timeout. - Conflicts resolve. Under concurrent writers on
REPEATABLE READ, serialization failures are retried and eventually succeed, with the retry count exported.
Pitfalls and edge cases¶
- External calls inside a transaction. Holds a connection and its locks for the entire remote call.
- Manual
BEGIN/COMMITviaexecute(). An exception leaves the pooled connection in an open transaction for the next borrower. - One connection shared across tasks. asyncpg connections are not concurrency-safe; acquire one per task.
prepare()results outliving the connection. Prepared statements are per connection; re-acquiring invalidates them.- Retrying individual statements. An aborted transaction rejects everything until rollback; retry the whole block.
- No
idle_in_transaction_session_timeout. A cancelled task at the wrong moment can leave locks held until the pool recycles. - Long-running batch inside one transaction. Holds a connection for minutes and bloats the write-ahead log; chunk it.
- Assuming
command_timeoutprotects the server. It cancels the client wait; the server-sidestatement_timeoutstops the work.
Frequently Asked Questions¶
Should an HTTP call ever happen inside a database transaction?
No. The connection and every lock the transaction holds stay checked out for the whole remote call, so a 400 ms API call turns into 400 ms of pool and lock occupancy per request. Read what you need, release the connection, make the call, then re-acquire for the write — and if atomicity is required, use an optimistic version check in the update instead.
How do savepoints work in asyncpg?
A nested async with conn.transaction() block issues a SAVEPOINT rather than a new transaction, so an exception inside it rolls back only that block and leaves the outer transaction usable. This is what lets a batch import skip a single bad row instead of aborting everything, at the cost of one extra round trip per savepoint.
What happens to an asyncpg transaction if the task is cancelled?
The context manager's exit path issues a rollback, so a cancelled task normally leaves no open transaction. The risk is cancellation arriving while a statement is in flight and the rollback itself being interrupted; set idle_in_transaction_session_timeout on the server so PostgreSQL terminates any session that is left holding locks.
Which database errors are safe to retry?
Serialization failures (SQLSTATE 40001) and deadlocks (40P01), and only by re-running the entire transaction with jittered backoff. Constraint violations, syntax errors and permission failures are deterministic and will fail identically on retry. Never retry individual statements inside an aborted transaction — PostgreSQL rejects everything until the rollback completes.
Related¶
- Async Database Drivers — the parent overview: drivers, pools, and blocking pitfalls.
- SQLAlchemy asyncio session-per-request patterns — the ORM layer above these transactions.
- Retry & Backoff Strategies — the jittered backoff used for conflict retries.