2 Commits
Author SHA1 Message Date
dsql 3258020fa7 chore: bump to 1.1.1
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-10-16 04:32:47 -04:00
dsql 2fb078ac6f 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>
2026-10-16 04:32:47 -04:00
3 changed files with 16 additions and 34 deletions

No files matched your search

+3 -3
View File
@@ -10,18 +10,18 @@ a sibling of the `mongo` lib. Class is **`PsqlDB`**.
`requirements.txt`: `requirements.txt`:
``` ```
psql @ git+ssh://git@git.rethinkstudios.io/rethink-public/psql.git@v1.0.0 psql @ git+ssh://git@git.rethinkstudios.io/rethink-public/psql.git@v1.1.1
``` ```
Direct: Direct:
```bash ```bash
pip install "psql @ git+ssh://git@git.rethinkstudios.io/rethink-public/psql.git@v1.0.0" pip install "psql @ git+ssh://git@git.rethinkstudios.io/rethink-public/psql.git@v1.1.1"
``` ```
Pulls `asyncpg`. Pulls `asyncpg`.
Drop the `@v1.0.0` suffix from the line above to install the latest unpinned. Drop the `@v1.1.1` suffix from the line above to install the latest unpinned.
## The two-layer API ## The two-layer API
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "psql" name = "psql"
version = "1.1.0" version = "1.1.1"
description = "async postgres wrapper over asyncpg: two-layer API (friendly verbs + raw escape hatch), fail-loud, config-free" description = "async postgres wrapper over asyncpg: two-layer API (friendly verbs + raw escape hatch), fail-loud, config-free"
requires-python = ">=3.10" requires-python = ">=3.10"
dependencies = [ dependencies = [
+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: