8 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
dsql 71993fea1b docs: add __aenter__/__aexit__ one-liner docstrings for twin parity with redis
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-06 02:55:58 -04:00
dsql e4c61cf6c3 fix: log-and-reraise create_pool failures; lock close() against a racing connect()
connect() built the pool with `await asyncpg.create_pool(...)` outside the
try that wraps SELECT-1 validation - with the default min_size=1 the pool
connects eagerly, so a bad host/auth propagated with no log.exception line,
breaking the documented "every method logs then re-raises" contract.
create_pool() now sits inside the same try/log/re-raise as validation, and
either failure point tears down a partially-built pool before re-raising.

close() didn't take _connect_lock while connect() did, so a close() racing
an in-flight connect() would see _pool is None and no-op as success while
connect() went on to install a live pool - a shutdown handler racing a
reconnect could "close" the instance while real connections stayed open.
close() now takes the same lock via a shared _close_locked() helper.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-03 19:11:52 -04:00
dsql d619d4ca55 refactor: derive __version__ from package metadata (single source)
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-03 17:00:59 -04:00
dsql f07a8da5a4 docs: compress prose/module docstrings, em-dash->hyphen (de-bloat wave 1)
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-03 00:16:11 -04:00
dsql f90c18ed64 fix: close() nulls pool + connect() lock guards against concurrent races (psql-7, psql-8)
- close() sets self._pool = None (in a finally, even on driver-error close) so a closed
  instance reports not-connected instead of masquerading as still-connected against a
  dead pool (the pool property now raises RuntimeError post-close as intended).
- connect() is now guarded by an internal asyncio.Lock: concurrent connect() calls
  serialize instead of each racing to build + orphan its own live pool. Twin fix with
  mysql at the same Layer-1 signature level.
- docstrings tightened (module header lifecycle/dsn notes, __init__ dsn note) with zero
  behavior change; re-verified against a real embedded postgres.

verified against pixeltable-pgserver (test-only): 8 concurrent connect() calls now leave
0 orphaned pools + exactly 1 live pool (old: 7 orphaned, 8 live simultaneously); close()
now flips the connected-check to False (old: stayed True against a dead pool). full
layer-1/layer-2/transaction/concurrency regression suite still green. fresh-venv install
confirms runtime purity (asyncpg only, no pixeltable-pgserver). bump v0.1.4 -> v0.1.5

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-02 23:25:52 -04:00
dsql 1ae280e5c3 fix: host/port defaults no longer shadow a dsn's embedded host/port (psql-5)
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-02 17:30:10 -04:00
dsql c3205f0614 fix: _where() renders None conditions as IS NULL instead of = NULL
col = $n bound to NULL never matches in sql, so get/get_one/exists/delete
silently missed every row filtered on a None value despite insert() writing
NULL fine. Kept in lockstep with the mysql lib's identical fix.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-02 16:44:27 -04:00
4 changed files with 110 additions and 65 deletions
+9 -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@v0.1.2 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@v0.1.2" pip install "psql @ git+ssh://git@git.rethinkstudios.io/rethink-public/psql.git@v0.1.7"
``` ```
Pulls `asyncpg`. Pulls `asyncpg`.
Drop the `@v0.1.2` 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
@@ -63,6 +63,12 @@ async with PsqlDB(database="app", user="postgres") as db:
await db.insert("events", {"kind": "login"}) await db.insert("events", {"kind": "login"})
``` ```
`host`/`port` default to `None`, not `"localhost"`/`5432` — asyncpg only reads a `dsn`'s
embedded host/port when the `host`/`port` kwargs are falsy, so passing `dsn=...` in
`pool_kwargs` (with no `host`/`port` of your own) lets the dsn's server reach asyncpg
instead of being silently overridden. The no-dsn path above still defaults to
`localhost:5432` when you don't pass `host`.
### Layer 2 — raw SQL for the complex queries ### Layer 2 — raw SQL for the complex queries
```python ```python
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "psql" name = "psql"
version = "0.1.2" 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 = [
+7
View File
@@ -1,3 +1,10 @@
from importlib.metadata import PackageNotFoundError, version
from .psql import PsqlDB from .psql import PsqlDB
__all__ = ["PsqlDB"] __all__ = ["PsqlDB"]
try:
__version__ = version("psql")
except PackageNotFoundError:
__version__ = "0.0.0+unknown"
+82 -50
View File
@@ -1,5 +1,5 @@
""" """
async postgres wrapper over asyncpg two-layer API (friendly verbs + raw escape hatch) async postgres wrapper over asyncpg - two-layer API (friendly verbs + raw escape hatch)
object pattern (one pool per process), attach to the app: object pattern (one pool per process), attach to the app:
from psql import PsqlDB from psql import PsqlDB
@@ -13,39 +13,33 @@ context manager:
async with PsqlDB(database="app", user="postgres") as db: async with PsqlDB(database="app", user="postgres") as db:
await db.execute("CREATE TABLE ...") await db.execute("CREATE TABLE ...")
lifecycle: dsn:
construction is sync and opens no socket. connect() builds the asyncpg pool and host/port default to None, not "localhost"/5432 - asyncpg only reads a dsn's embedded
validates it with `SELECT 1` so a bad host/credentials fails loud immediately rather host/port when the host/port kwargs are falsy, so a hardcoded default would silently
than on the first real op, and returns self. close() closes the pool. shadow the dsn's server. pass dsn=... in pool_kwargs alone (no host/port) to let the
dsn's own host/port reach asyncpg; the no-dsn path still defaults to localhost:5432.
two-layer API: two-layer API:
LAYER 1 friendly, portable verbs for simple single-table CRUD. these hide the LAYER 1 - friendly, portable verbs for simple single-table CRUD, byte-for-byte
dialect and are byte-for-byte identical to the `mysql` lib, so a dev swaps psql<->mysql identical to the `mysql` lib (zero call-site changes swapping psql<->mysql):
with zero call-site changes: create_database, create_table, drop, insert, get, create_database, create_table, drop, insert, get, get_one, delete, exists, upsert.
get_one, delete, exists, upsert. LAYER 2 - raw escape hatch for the complex ~20% (joins, aggregates, CTEs, window
LAYER 2 — raw escape hatch for the complex ~20% (joins, aggregates, CTEs, window functions): execute, fetch, fetchone, fetchval, transaction, using `$1, $2`
functions): execute, fetch, fetchone, fetchval, transaction. you write the SQL with placeholders (asyncpg style). deliberately nothing in between - no query builder/ORM.
`$1, $2` placeholders (asyncpg style) + params; the wrapper still gives pooling,
parameterization, fail-loud errors, and row->dict conversion. raw SQL is psql-specific.
there is deliberately NOTHING in between — no query builder / ORM. a join goes through
raw fetch(), never a chainable .where()/.join().
rows:
layer 1 and fetch/fetchone return plain dicts ({column: value}), not asyncpg Record
objects — identical shape to the mysql lib.
placeholders / injection safety: placeholders / injection safety:
values are ALWAYS parameterized layer 1 builds `$1, $2` internally; layer 2 takes values are ALWAYS parameterized - layer 1 builds `$1, $2` internally; layer 2 takes
your `$1` placeholders + *params. never f-string/format a value into SQL. only your `$1` placeholders + *params. never f-string/format a value into SQL. only
identifiers (table/column names) are interpolated, and they are quoted. identifiers (table/column names) are interpolated, and they are quoted.
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), logs via getLogger(__name__), and re-raises. a None/[] return is
only ever a real result (no row, empty table) never a swallowed failure. for anything 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 logging import logging
from typing import Any, Dict, List, Optional, Sequence from typing import Any, Dict, List, Optional, Sequence
@@ -59,7 +53,7 @@ _DRIVER_ERRORS = (asyncpg.PostgresError, asyncpg.InterfaceError, OSError)
def _quote_ident(identifier: str) -> str: def _quote_ident(identifier: str) -> str:
"""quote a sql identifier (table/column), escaping embedded double-quotes """quote a sql identifier (table/column), escaping embedded double-quotes
identifiers can't be parameterized, so they are interpolated quoting + doubling any identifiers can't be parameterized, so they are interpolated - quoting + doubling any
embedded quote is the postgres-safe way to do that for caller-supplied names. embedded quote is the postgres-safe way to do that for caller-supplied names.
""" """
return '"' + identifier.replace('"', '""') + '"' return '"' + identifier.replace('"', '""') + '"'
@@ -75,8 +69,8 @@ class PsqlDB:
def __init__( def __init__(
self, self,
host: str = "localhost", host: Optional[str] = None,
port: int = 5432, port: Optional[int] = None,
database: Optional[str] = None, database: Optional[str] = None,
user: Optional[str] = None, user: Optional[str] = None,
password: Optional[str] = None, password: Optional[str] = None,
@@ -91,7 +85,18 @@ class PsqlDB:
host/port/database/user/password/min_size/max_size/command_timeout are injected by host/port/database/user/password/min_size/max_size/command_timeout are injected by
the caller. extra pool_kwargs pass through to asyncpg.create_pool (ssl, server_ the caller. extra pool_kwargs pass through to asyncpg.create_pool (ssl, server_
settings, dsn, etc). `host` may be a unix socket directory as well as a hostname. settings, dsn, etc). `host` may be a unix socket directory as well as a hostname.
host/port default to None here (not "localhost"/5432): asyncpg only reads a dsn's
embedded host/port when the host/port kwargs are falsy, so a hardcoded default
would silently shadow the dsn's server. no dsn -> host/port fall back to
localhost:5432; dsn passed -> host/port stay None (letting the dsn's own reach
asyncpg) unless the caller explicitly overrides them.
""" """
if "dsn" not in pool_kwargs:
if host is None:
host = "localhost"
if port is None:
port = 5432
self._config = dict( self._config = dict(
host=host, host=host,
port=port, port=port,
@@ -104,6 +109,7 @@ class PsqlDB:
**pool_kwargs, **pool_kwargs,
) )
self._pool: Optional[asyncpg.Pool] = None self._pool: Optional[asyncpg.Pool] = None
self._connect_lock = asyncio.Lock()
async def connect(self) -> "PsqlDB": async def connect(self) -> "PsqlDB":
"""build the pool and validate it with SELECT 1; fail loud on bad config """build the pool and validate it with SELECT 1; fail loud on bad config
@@ -111,25 +117,42 @@ class PsqlDB:
returns self so callers can write `db = await PsqlDB(...).connect()`. if called returns self so callers can write `db = await PsqlDB(...).connect()`. if called
again on an already-connected instance the previous pool is closed first (no again on an already-connected instance the previous pool is closed first (no
orphaned pool); if the SELECT-1 validation fails the freshly-built pool is torn orphaned pool); if the SELECT-1 validation fails the freshly-built pool is torn
down before re-raising, so a failed connect() never leaks a live pool. down before re-raising, so a failed connect() never leaks a live pool. guarded by
an internal lock so concurrent connect() calls build exactly one pool instead of
each racing to create + orphan their own.
""" """
async with self._connect_lock:
if self._pool is not None: if self._pool is not None:
await self.close() await self._close_locked()
pool = await asyncpg.create_pool(**self._config) pool = None
try: try:
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() validation failed") log.exception("psql.connect() failed")
if pool is not None:
await pool.close() await pool.close()
raise raise
except BaseException: except BaseException:
if pool is not None:
await pool.close() await pool.close()
raise raise
self._pool = pool self._pool = pool
return self return self
async def close(self) -> None: async def close(self) -> None:
"""close the pool on shutdown""" """close the pool on shutdown and null the reference (so pool/connected checks
report not-connected against a dead pool)
guarded by the same lock as connect() - a close() racing an in-flight connect()
waits for it rather than no-opping against a not-yet-installed pool and leaving
the just-built one live.
"""
async with self._connect_lock:
await self._close_locked()
async def _close_locked(self) -> None:
"""close the pool and null the reference; caller must hold `_connect_lock`"""
if self._pool is None: if self._pool is None:
return return
try: try:
@@ -137,31 +160,33 @@ class PsqlDB:
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("psql.close()") log.exception("psql.close()")
raise raise
finally:
self._pool = None
async def __aenter__(self) -> "PsqlDB": async def __aenter__(self) -> "PsqlDB":
"""enter: connect() and return self"""
return await self.connect() return await self.connect()
async def __aexit__(self, exc_type, exc, tb) -> None: async def __aexit__(self, exc_type, exc, tb) -> None:
"""exit: close(), ignoring exc_type/exc/tb"""
await self.close() await self.close()
@property @property
def pool(self) -> asyncpg.Pool: def pool(self) -> asyncpg.Pool:
"""raw asyncpg.Pool escape hatch; full driver surface, raises """raw asyncpg.Pool escape hatch for copy/prepare/listen-notify/cursors and
anything not wrapped; full driver surface, raises"""
use for copy/prepare/listen-notify/cursors and anything not wrapped.
"""
if self._pool is None: if self._pool is None:
raise RuntimeError("psql: not connected; call await db.connect() first") raise RuntimeError("psql: not connected; call await db.connect() first")
return self._pool return self._pool
# ------------------------------------------------------------------------- # -------------------------------------------------------------------------
# layer 2 raw escape hatch (you write the SQL, $1 placeholders) # layer 2 - raw escape hatch (you write the SQL, $1 placeholders)
async def execute(self, query: str, *params: Any) -> str: async def execute(self, query: str, *params: Any) -> str:
"""run a statement (INSERT/UPDATE/DELETE/DDL); return asyncpg's status string """run a statement (INSERT/UPDATE/DELETE/DDL); return asyncpg's status string
the status string is e.g. "INSERT 0 1" / "UPDATE 3" / "DELETE 2" parse it or use e.g. "INSERT 0 1" / "UPDATE 3" / "DELETE 2" - parse it, or use the layer-1 verbs
the layer-1 verbs (insert/delete) which return structured values instead. (insert/delete) which return structured values instead.
""" """
try: try:
return await self.pool.execute(query, *params) return await self.pool.execute(query, *params)
@@ -181,8 +206,7 @@ class PsqlDB:
async def fetchone(self, query: str, *params: Any) -> Optional[dict]: async def fetchone(self, query: str, *params: Any) -> Optional[dict]:
"""run a query and return the first row as a dict, or None if no rows """run a query and return the first row as a dict, or None if no rows
named fetchone (not asyncpg's fetchrow) to match the mysql lib's layer-2 surface; named fetchone (not asyncpg's fetchrow) to match the mysql lib's layer-2 surface.
maps to the driver's fetchrow internally.
""" """
try: try:
row = await self.pool.fetchrow(query, *params) row = await self.pool.fetchrow(query, *params)
@@ -207,15 +231,15 @@ class PsqlDB:
await conn.execute("INSERT ...", a) await conn.execute("INSERT ...", a)
await conn.execute("UPDATE ...", b) await conn.execute("UPDATE ...", b)
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 (use its $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)
async def create_database(self, name: str) -> None: async def create_database(self, name: str) -> None:
"""CREATE DATABASE name (raises if it already exists postgres has no IF NOT """CREATE DATABASE name (raises if it already exists - postgres has no IF NOT
EXISTS for CREATE DATABASE; catch the duplicate error or check first)""" EXISTS for CREATE DATABASE; catch the duplicate error or check first)"""
try: try:
await self.pool.execute(f"CREATE DATABASE {_quote_ident(name)}") await self.pool.execute(f"CREATE DATABASE {_quote_ident(name)}")
@@ -251,7 +275,7 @@ class PsqlDB:
"""INSERT one row from {column: value}; return the inserted rowcount (1) """INSERT one row from {column: value}; return the inserted rowcount (1)
values are parameterized ($1, $2, ...). returns the number of rows inserted (1 on values are parameterized ($1, $2, ...). returns the number of rows inserted (1 on
success) the portable return shape shared with the mysql lib. success) - the portable return shape shared with the mysql lib.
""" """
cols = list(values.keys()) cols = list(values.keys())
placeholders = ", ".join(f"${i + 1}" for i in range(len(cols))) placeholders = ", ".join(f"${i + 1}" for i in range(len(cols)))
@@ -267,7 +291,7 @@ class PsqlDB:
async def get(self, table: str, conditions: Optional[Dict[str, Any]] = None) -> List[dict]: async def get(self, table: str, conditions: Optional[Dict[str, Any]] = None) -> List[dict]:
"""SELECT * rows matching equality `conditions` (col = val AND ...) as dicts """SELECT * rows matching equality `conditions` (col = val AND ...) as dicts
conditions=None/{} returns all rows. simple equality only anything more complex conditions=None/{} returns all rows. simple equality only - anything more complex
goes through raw fetch(). goes through raw fetch().
""" """
where, params = _where(conditions) where, params = _where(conditions)
@@ -315,7 +339,7 @@ class PsqlDB:
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:
"""INSERT ... ON CONFLICT (conflict_cols) DO UPDATE insert or update on key clash """INSERT ... ON CONFLICT (conflict_cols) DO UPDATE - insert or update on key clash
`conflict` is the list of columns forming the unique/pk constraint to upsert on. `conflict` is the list of columns forming the unique/pk constraint to upsert on.
the wrapper emits ON CONFLICT here (the mysql lib emits ON DUPLICATE KEY UPDATE for the wrapper emits ON CONFLICT here (the mysql lib emits ON DUPLICATE KEY UPDATE for
@@ -353,7 +377,7 @@ class _Transaction:
await self._tx.start() await self._tx.start()
except BaseException: except BaseException:
# start() (or transaction()) failing after acquire would otherwise leak the # start() (or transaction()) failing after acquire would otherwise leak the
# pooled connection __aexit__ is not called when __aenter__ raises. release # pooled connection - __aexit__ is not called when __aenter__ raises. release
# it and reset so a failed transaction start never burns a pool slot. # it and reset so a failed transaction start never burns a pool slot.
await self._pool.release(self._conn) await self._pool.release(self._conn)
self._conn = None self._conn = None
@@ -373,13 +397,21 @@ class _Transaction:
def _where(conditions: Optional[Dict[str, Any]]) -> tuple: def _where(conditions: Optional[Dict[str, Any]]) -> tuple:
"""build a parameterized `WHERE col = $1 AND ...` clause + the params list """build a parameterized `WHERE col = $1 AND ...` clause + the params list
returns ("", []) when there are no conditions. equality only. returns ("", []) when there are no conditions. equality only. a None value renders as
`col IS NULL` (not `col = $n` bound to NULL, which sql never matches) and does not
consume a placeholder.
""" """
if not conditions: if not conditions:
return "", [] return "", []
cols = list(conditions.keys()) parts = []
clause = " AND ".join(f"{_quote_ident(c)} = ${i + 1}" for i, c in enumerate(cols)) params = []
return f" WHERE {clause}", list(conditions.values()) for col, val in conditions.items():
if val is None:
parts.append(f"{_quote_ident(col)} IS NULL")
else:
params.append(val)
parts.append(f"{_quote_ident(col)} = ${len(params)}")
return f" WHERE {' AND '.join(parts)}", params
def _status_count(status: str) -> int: def _status_count(status: str) -> int: