refactor: transaction() as an asynccontextmanager over aiomysql's acquire context

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 7eb5d552f5
commit dc9ab8a45c
1 file changed
+13 -29
+13 -29
View File
@@ -26,6 +26,7 @@ for anything not wrapped, use `.pool`.
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 aiomysql import aiomysql
@@ -271,7 +272,7 @@ class MysqlDB:
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
aiomysql connection (autocommit disabled for the block). aiomysql connection (autocommit disabled for the block).
""" """
return _Transaction(self.pool) return _transaction(self.pool)
# ------------------------------------------------------------------------- # -------------------------------------------------------------------------
# layer 1 - friendly portable verbs (identical to the psql lib) # layer 1 - friendly portable verbs (identical to the psql lib)
@@ -400,39 +401,22 @@ class MysqlDB:
return rowcount return rowcount
class _Transaction: @asynccontextmanager
"""async ctx mgr: acquire a pooled connection + run a transaction on it async def _transaction(pool: aiomysql.Pool):
"""acquire a pooled connection and run the block in a transaction on it
autocommit is disabled for the block so the statements commit together on clean begin() disables autocommit for the block; commit on clean exit, rollback + re-raise
exit (or roll back on error). on any error. the acquire context releases the connection on every path - including
a failed begin(), so a failed start never burns a pool slot
""" """
async with pool.acquire() as conn:
def __init__(self, pool: aiomysql.Pool): await conn.begin()
self._pool = pool
self._conn = None
async def __aenter__(self):
self._conn = await self._pool.acquire()
try: try:
await self._conn.begin() yield conn
except BaseException: except BaseException:
# __aexit__ never runs if __aenter__ raises, so begin() failing after acquire await conn.rollback()
# would otherwise leak the pooled connection - release it here instead.
await self._pool.release(self._conn)
self._conn = None
raise raise
return self._conn await conn.commit()
async def __aexit__(self, exc_type, exc, tb) -> None:
try:
if exc_type is None:
await self._conn.commit()
else:
await self._conn.rollback()
finally:
# release() returns a task (schedules a _wakeup); await it to avoid leaving
# it un-awaited each transaction, matching psql's and aiomysql's own idiom.
await self._pool.release(self._conn)
def _where(conditions: Optional[Dict[str, Any]]) -> tuple: def _where(conditions: Optional[Dict[str, Any]]) -> tuple: