1 Commits
Author SHA1 Message Date
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
3 changed files with 42 additions and 34 deletions
+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@v0.1.4 psql @ git+ssh://git@git.rethinkstudios.io/rethink-public/psql.git@v0.1.5
``` ```
Direct: Direct:
```bash ```bash
pip install "psql @ git+ssh://git@git.rethinkstudios.io/rethink-public/psql.git@v0.1.4" pip install "psql @ git+ssh://git@git.rethinkstudios.io/rethink-public/psql.git@v0.1.5"
``` ```
Pulls `asyncpg`. Pulls `asyncpg`.
Drop the `@v0.1.4` suffix from the line above to install the latest unpinned. Drop the `@v0.1.5` 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 = "0.1.4" version = "0.1.5"
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 = [
+38 -30
View File
@@ -14,16 +14,17 @@ context manager:
await db.execute("CREATE TABLE ...") await db.execute("CREATE TABLE ...")
lifecycle: lifecycle:
construction is sync and opens no socket. connect() builds the asyncpg pool and construction is sync, opens no socket. connect() builds the asyncpg pool, validates it
validates it with `SELECT 1` so a bad host/credentials fails loud immediately rather with `SELECT 1` (fail loud on bad host/credentials immediately, not on the first real
than on the first real op, and returns self. close() closes the pool. op), returns self; concurrent connect() calls are lock-serialized so only one pool is
ever live. close() closes the pool and nulls the reference (a closed instance reports
not-connected).
dsn: dsn:
host/port default to None, not "localhost"/5432 — asyncpg only reads a dsn's host/port default to None, not "localhost"/5432 — asyncpg only reads a dsn's embedded
embedded host/port when the host/port kwargs are falsy, so a hardcoded default would host/port when the host/port kwargs are falsy, so a hardcoded default would silently
silently shadow the dsn's server. pass dsn=... in pool_kwargs alone (no host/port) to shadow the dsn's server. pass dsn=... in pool_kwargs alone (no host/port) to let the
let the dsn's own host/port reach asyncpg; the no-dsn path still defaults to dsn's own host/port reach asyncpg; the no-dsn path still defaults to localhost:5432.
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. these hide the
@@ -53,6 +54,7 @@ errors (FAIL LOUD — unlike the mongo lib's swallow-and-default):
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
@@ -99,12 +101,11 @@ class PsqlDB:
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) because asyncpg only reads host/port default to None here (not "localhost"/5432): asyncpg only reads a dsn's
a dsn's embedded host/port when the host/port kwargs are falsy a hardcoded embedded host/port when the host/port kwargs are falsy, so a hardcoded default
default would silently shadow the dsn's server and connect you to the wrong one. would silently shadow the dsn's server. no dsn -> host/port fall back to
when no dsn is passed, host/port fall back to localhost:5432 (the common no-dsn localhost:5432; dsn passed -> host/port stay None (letting the dsn's own reach
path is unchanged); when a dsn is passed, host/port stay None unless the caller asyncpg) unless the caller explicitly overrides them.
explicitly overrides them, letting the dsn's own host/port reach asyncpg.
""" """
if "dsn" not in pool_kwargs: if "dsn" not in pool_kwargs:
if host is None: if host is None:
@@ -123,6 +124,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
@@ -130,25 +132,29 @@ 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.
""" """
if self._pool is not None: async with self._connect_lock:
await self.close() if self._pool is not None:
pool = await asyncpg.create_pool(**self._config) await self.close()
try: pool = await asyncpg.create_pool(**self._config)
await pool.fetchval("SELECT 1") try:
except _DRIVER_ERRORS: await pool.fetchval("SELECT 1")
log.exception("psql.connect() validation failed") except _DRIVER_ERRORS:
await pool.close() log.exception("psql.connect() validation failed")
raise await pool.close()
except BaseException: raise
await pool.close() except BaseException:
raise await pool.close()
self._pool = pool raise
return self self._pool = pool
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)"""
if self._pool is None: if self._pool is None:
return return
try: try:
@@ -156,6 +162,8 @@ 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":
return await self.connect() return await self.connect()