4 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
dsql 6466b797d3 chore: bump to 1.1.0 (logging-discipline audit)
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-08-10 23:00:50 -04:00
dsql 7691c79842 fix: log-XOR-raise — demote the pre-raise log to DEBUG, revise the contract (D1)
every wrapped method did log.exception (ERROR + traceback) immediately before re-raising
the driver error — the same failure reported at two layers (the log AND the raised
exception). that is the log-and-raise violation: a method that re-raises must not also
error-log, because the caller — which alone knows whether the failure is fatal or routine
— is the one that logs. demote all wrapped-method logs to log.debug(..., exc_info=True):
the traceback stays available at DEBUG, and the raised exception is the single loud
terminal signal. docstring updated from "logs via getLogger and re-raises" to the
corrected raise-XOR-log contract. no behavior change beyond log level — the fail-loud
re-raise is unchanged.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-08-09 02:18:08 -04:00
3 changed files with 36 additions and 51 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.0.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 = [
+32 -47
View File
@@ -34,13 +34,17 @@ placeholders / injection safety:
errors (FAIL LOUD - unlike the mongo lib's swallow-and-default): errors (FAIL LOUD - unlike the mongo lib's swallow-and-default):
every method catches the driver error (asyncpg.PostgresError / InterfaceError, OSError every method catches the driver error (asyncpg.PostgresError / InterfaceError, OSError
on connection loss), logs via getLogger(__name__), and re-raises. a None/[] return is on connection loss) and re-raises it - the raised exception IS the signal, and the
only ever a real result (no row, empty table) - never a swallowed failure. for anything caller (which alone knows fatal-vs-routine) decides and logs. the wrapped method itself
logs only at DEBUG (with the traceback) so a failure isn't reported twice (raise XOR
log). a None/[] return is only ever a real result (no row, empty table) - never a
swallowed failure. for anything
not wrapped, use the raw `.pool` property (the asyncpg.Pool). not wrapped, use the raw `.pool` property (the asyncpg.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 asyncpg import asyncpg
@@ -129,7 +133,7 @@ class PsqlDB:
pool = await asyncpg.create_pool(**self._config) pool = await asyncpg.create_pool(**self._config)
await pool.fetchval("SELECT 1") await pool.fetchval("SELECT 1")
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.connect() failed") log.debug("psql.connect() failed", exc_info=True)
if pool is not None: if pool is not None:
await pool.close() await pool.close()
raise raise
@@ -158,7 +162,7 @@ class PsqlDB:
try: try:
await self._pool.close() await self._pool.close()
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.close()") log.debug("psql.close()", exc_info=True)
raise raise
finally: finally:
self._pool = None self._pool = None
@@ -191,7 +195,7 @@ class PsqlDB:
try: try:
return await self.pool.execute(query, *params) return await self.pool.execute(query, *params)
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.execute(): %s", query) log.debug("psql.execute(): %s", query, exc_info=True)
raise raise
async def fetch(self, query: str, *params: Any) -> List[dict]: async def fetch(self, query: str, *params: Any) -> List[dict]:
@@ -199,7 +203,7 @@ class PsqlDB:
try: try:
rows = await self.pool.fetch(query, *params) rows = await self.pool.fetch(query, *params)
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.fetch(): %s", query) log.debug("psql.fetch(): %s", query, exc_info=True)
raise raise
return [_row_to_dict(r) for r in rows] return [_row_to_dict(r) for r in rows]
@@ -211,7 +215,7 @@ class PsqlDB:
try: try:
row = await self.pool.fetchrow(query, *params) row = await self.pool.fetchrow(query, *params)
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.fetchone(): %s", query) log.debug("psql.fetchone(): %s", query, exc_info=True)
raise raise
return _row_to_dict(row) if row is not None else None return _row_to_dict(row) if row is not None else None
@@ -220,7 +224,7 @@ class PsqlDB:
try: try:
return await self.pool.fetchval(query, *params) return await self.pool.fetchval(query, *params)
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.fetchval(): %s", query) log.debug("psql.fetchval(): %s", query, exc_info=True)
raise raise
def transaction(self): def transaction(self):
@@ -233,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)
@@ -244,7 +248,7 @@ class PsqlDB:
try: try:
await self.pool.execute(f"CREATE DATABASE {_quote_ident(name)}") await self.pool.execute(f"CREATE DATABASE {_quote_ident(name)}")
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.create_database(%s)", name) log.debug("psql.create_database(%s)", name, exc_info=True)
raise raise
async def create_table(self, name: str, schema: Dict[str, str]) -> None: async def create_table(self, name: str, schema: Dict[str, str]) -> None:
@@ -259,7 +263,7 @@ class PsqlDB:
try: try:
await self.pool.execute(f"CREATE TABLE IF NOT EXISTS {_quote_ident(name)} ({cols})") await self.pool.execute(f"CREATE TABLE IF NOT EXISTS {_quote_ident(name)} ({cols})")
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.create_table(%s)", name) log.debug("psql.create_table(%s)", name, exc_info=True)
raise raise
async def drop(self, name: str, *, table: bool = True) -> None: async def drop(self, name: str, *, table: bool = True) -> None:
@@ -268,7 +272,7 @@ class PsqlDB:
try: try:
await self.pool.execute(f"DROP {kind} IF EXISTS {_quote_ident(name)}") await self.pool.execute(f"DROP {kind} IF EXISTS {_quote_ident(name)}")
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.drop(%s)", name) log.debug("psql.drop(%s)", name, exc_info=True)
raise raise
async def insert(self, table: str, values: Dict[str, Any]) -> int: async def insert(self, table: str, values: Dict[str, Any]) -> int:
@@ -284,7 +288,7 @@ class PsqlDB:
try: try:
status = await self.pool.execute(query, *values.values()) status = await self.pool.execute(query, *values.values())
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.insert(%s)", table) log.debug("psql.insert(%s)", table, exc_info=True)
raise raise
return _status_count(status) return _status_count(status)
@@ -299,7 +303,7 @@ class PsqlDB:
try: try:
rows = await self.pool.fetch(query, *params) rows = await self.pool.fetch(query, *params)
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.get(%s)", table) log.debug("psql.get(%s)", table, exc_info=True)
raise raise
return [_row_to_dict(r) for r in rows] return [_row_to_dict(r) for r in rows]
@@ -310,7 +314,7 @@ class PsqlDB:
try: try:
row = await self.pool.fetchrow(query, *params) row = await self.pool.fetchrow(query, *params)
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.get_one(%s)", table) log.debug("psql.get_one(%s)", table, exc_info=True)
raise raise
return _row_to_dict(row) if row is not None else None return _row_to_dict(row) if row is not None else None
@@ -324,7 +328,7 @@ class PsqlDB:
try: try:
status = await self.pool.execute(query, *params) status = await self.pool.execute(query, *params)
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.delete(%s)", table) log.debug("psql.delete(%s)", table, exc_info=True)
raise raise
return _status_count(status) return _status_count(status)
@@ -335,7 +339,7 @@ class PsqlDB:
try: try:
return bool(await self.pool.fetchval(query, *params)) return bool(await self.pool.fetchval(query, *params))
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.exists(%s)", table) log.debug("psql.exists(%s)", table, exc_info=True)
raise raise
async def upsert(self, table: str, values: Dict[str, Any], conflict: Sequence[str]) -> int: async def upsert(self, table: str, values: Dict[str, Any], conflict: Sequence[str]) -> int:
@@ -357,41 +361,22 @@ class PsqlDB:
try: try:
status = await self.pool.execute(query, *values.values()) status = await self.pool.execute(query, *values.values())
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.upsert(%s)", table) log.debug("psql.upsert(%s)", table, exc_info=True)
raise raise
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: