1 Commits
Author SHA1 Message Date
dsql 49e8de3b1a release: 1.0.0
first stable release. pre-1.0.0 verification complete: all surviving MED regressions and
gaps resolved and independently re-fired, tree audited clean across the suite.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-06 21:21:00 -04:00
3 changed files with 51 additions and 36 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.1.1 psql @ git+ssh://git@git.rethinkstudios.io/rethink-public/psql.git@v0.1.7
``` ```
Direct: Direct:
```bash ```bash
pip install "psql @ git+ssh://git@git.rethinkstudios.io/rethink-public/psql.git@v1.1.1" pip install "psql @ git+ssh://git@git.rethinkstudios.io/rethink-public/psql.git@v0.1.7"
``` ```
Pulls `asyncpg`. Pulls `asyncpg`.
Drop the `@v1.1.1` suffix from the line above to install the latest unpinned. Drop the `@v0.1.7` 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.1" version = "1.0.0"
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 = [
+47 -32
View File
@@ -34,17 +34,13 @@ 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) and re-raises it - the raised exception IS the signal, and the on connection loss), logs via getLogger(__name__), and re-raises. a None/[] return is
caller (which alone knows fatal-vs-routine) decides and logs. the wrapped method itself only ever a real result (no row, empty table) - never a swallowed failure. for anything
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
@@ -133,7 +129,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.debug("psql.connect() failed", exc_info=True) log.exception("psql.connect() failed")
if pool is not None: if pool is not None:
await pool.close() await pool.close()
raise raise
@@ -162,7 +158,7 @@ class PsqlDB:
try: try:
await self._pool.close() await self._pool.close()
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.debug("psql.close()", exc_info=True) log.exception("psql.close()")
raise raise
finally: finally:
self._pool = None self._pool = None
@@ -195,7 +191,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.debug("psql.execute(): %s", query, exc_info=True) log.exception("psql.execute(): %s", query)
raise raise
async def fetch(self, query: str, *params: Any) -> List[dict]: async def fetch(self, query: str, *params: Any) -> List[dict]:
@@ -203,7 +199,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.debug("psql.fetch(): %s", query, exc_info=True) log.exception("psql.fetch(): %s", query)
raise raise
return [_row_to_dict(r) for r in rows] return [_row_to_dict(r) for r in rows]
@@ -215,7 +211,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.debug("psql.fetchone(): %s", query, exc_info=True) log.exception("psql.fetchone(): %s", query)
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
@@ -224,7 +220,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.debug("psql.fetchval(): %s", query, exc_info=True) log.exception("psql.fetchval(): %s", query)
raise raise
def transaction(self): def transaction(self):
@@ -237,7 +233,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)
@@ -248,7 +244,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.debug("psql.create_database(%s)", name, exc_info=True) log.exception("psql.create_database(%s)", name)
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:
@@ -263,7 +259,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.debug("psql.create_table(%s)", name, exc_info=True) log.exception("psql.create_table(%s)", name)
raise raise
async def drop(self, name: str, *, table: bool = True) -> None: async def drop(self, name: str, *, table: bool = True) -> None:
@@ -272,7 +268,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.debug("psql.drop(%s)", name, exc_info=True) log.exception("psql.drop(%s)", name)
raise raise
async def insert(self, table: str, values: Dict[str, Any]) -> int: async def insert(self, table: str, values: Dict[str, Any]) -> int:
@@ -288,7 +284,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.debug("psql.insert(%s)", table, exc_info=True) log.exception("psql.insert(%s)", table)
raise raise
return _status_count(status) return _status_count(status)
@@ -303,7 +299,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.debug("psql.get(%s)", table, exc_info=True) log.exception("psql.get(%s)", table)
raise raise
return [_row_to_dict(r) for r in rows] return [_row_to_dict(r) for r in rows]
@@ -314,7 +310,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.debug("psql.get_one(%s)", table, exc_info=True) log.exception("psql.get_one(%s)", table)
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
@@ -328,7 +324,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.debug("psql.delete(%s)", table, exc_info=True) log.exception("psql.delete(%s)", table)
raise raise
return _status_count(status) return _status_count(status)
@@ -339,7 +335,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.debug("psql.exists(%s)", table, exc_info=True) log.exception("psql.exists(%s)", table)
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:
@@ -361,22 +357,41 @@ 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.debug("psql.upsert(%s)", table, exc_info=True) log.exception("psql.upsert(%s)", table)
raise raise
return _status_count(status) return _status_count(status)
@asynccontextmanager class _Transaction:
async def _transaction(pool: asyncpg.Pool): """async ctx mgr: acquire a pooled connection + open a transaction on it"""
"""acquire a pooled connection and run the block in a transaction on it
asyncpg's own context managers do the work: the transaction commits on clean exit def __init__(self, pool: asyncpg.Pool):
and rolls back on any error, and the acquire releases the connection on every path - self._pool = pool
including a failed transaction start, so a failed start never burns a pool slot self._conn = None
""" self._tx = None
async with pool.acquire() as conn:
async with conn.transaction(): async def __aenter__(self):
yield conn self._conn = await self._pool.acquire()
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: