refactor: transaction() as an asynccontextmanager over asyncpg's own contexts

T3-21: the _Transaction class hand-rolled acquire / start / commit-or-rollback /
release plus a leak guard for a failed start. an asynccontextmanager over the
driver's own pool.acquire() context does the same, and the acquire context
releases on every path including a failed start. verified on a real backend:
commit, rollback + re-raise, cancellation rollback, failed-start leak guard
(pool max 1) and release after 50 transactions.

Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
dsql committed 2026-10-16 04:32:47 -04:00
1 parent 6466b797d3
commit 2fb078ac6f
1 file changed
+12 -30
+12 -30
View File
@@ -44,6 +44,7 @@ errors (FAIL LOUD - unlike the mongo lib's swallow-and-default):
import asyncio import asyncio
import logging import logging
from contextlib import asynccontextmanager
from typing import Any, Dict, List, Optional, Sequence from typing import Any, Dict, List, Optional, Sequence
import asyncpg import asyncpg
@@ -236,7 +237,7 @@ class PsqlDB:
commits on clean exit, rolls back and re-raises on any error. `conn` is a raw commits on clean exit, rolls back and re-raises on any error. `conn` is a raw
asyncpg connection ($1-placeholder execute/fetch/... directly). asyncpg connection ($1-placeholder execute/fetch/... directly).
""" """
return _Transaction(self.pool) return _transaction(self.pool)
# ------------------------------------------------------------------------- # -------------------------------------------------------------------------
# layer 1 - friendly portable verbs (identical across psql/mysql) # layer 1 - friendly portable verbs (identical across psql/mysql)
@@ -365,36 +366,17 @@ class PsqlDB:
return _status_count(status) return _status_count(status)
class _Transaction: @asynccontextmanager
"""async ctx mgr: acquire a pooled connection + open a transaction on it""" async def _transaction(pool: asyncpg.Pool):
"""acquire a pooled connection and run the block in a transaction on it
def __init__(self, pool: asyncpg.Pool): asyncpg's own context managers do the work: the transaction commits on clean exit
self._pool = pool and rolls back on any error, and the acquire releases the connection on every path -
self._conn = None including a failed transaction start, so a failed start never burns a pool slot
self._tx = None """
async with pool.acquire() as conn:
async def __aenter__(self): async with conn.transaction():
self._conn = await self._pool.acquire() yield conn
try:
self._tx = self._conn.transaction()
await self._tx.start()
except BaseException:
# start() (or transaction()) failing after acquire would otherwise leak the
# pooled connection - __aexit__ is not called when __aenter__ raises. release
# it and reset so a failed transaction start never burns a pool slot.
await self._pool.release(self._conn)
self._conn = None
raise
return self._conn
async def __aexit__(self, exc_type, exc, tb) -> None:
try:
if exc_type is None:
await self._tx.commit()
else:
await self._tx.rollback()
finally:
await self._pool.release(self._conn)
def _where(conditions: Optional[Dict[str, Any]]) -> tuple: def _where(conditions: Optional[Dict[str, Any]]) -> tuple: