add package: pyproject + src (layer-1 verbs + raw layer-2, fail-loud, plain-dict rows)

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-06-30 05:48:00 -04:00
parent cc3d6bb585
commit 919a019b63
3 changed files with 394 additions and 0 deletions
@@ -0,0 +1,15 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "psql"
+version = "0.1.0"
+description = "async postgres wrapper over asyncpg: two-layer API (friendly verbs + raw escape hatch), fail-loud, config-free"
+requires-python = ">=3.10"
+dependencies = [
+    "asyncpg>=0.29",
+]
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/psql"]
@@ -0,0 +1,3 @@
+from .psql import PsqlDB
+
+__all__ = ["PsqlDB"]
@@ -0,0 +1,376 @@
+"""
+async postgres wrapper over asyncpg — two-layer API (friendly verbs + raw escape hatch)
+
+object pattern (one pool per process), attach to the app:
+    from psql import PsqlDB
+    app.db = await PsqlDB(host="localhost", port=5432, database="app",
+                          user="postgres", password="secret").connect()
+    await app.db.insert("users", {"name": "ada", "active": True})
+    rows = await app.db.get("users", {"active": True})   # [{"name": "ada", ...}]
+    await app.db.close()                                  # on shutdown
+
+context manager:
+    async with PsqlDB(database="app", user="postgres") as db:
+        await db.execute("CREATE TABLE ...")
+
+lifecycle:
+    construction is sync and opens no socket. connect() builds the asyncpg pool and
+    validates it with `SELECT 1` so a bad host/credentials fails loud immediately rather
+    than on the first real op, and returns self. close() closes the pool.
+
+two-layer API:
+    LAYER 1 — friendly, portable verbs for simple single-table CRUD. these hide the
+    dialect and are byte-for-byte identical to the `mysql` lib, so a dev swaps psql<->mysql
+    with zero call-site changes: create_database, create_table, drop, insert, get,
+    get_one, delete, exists, upsert.
+    LAYER 2 — raw escape hatch for the complex ~20% (joins, aggregates, CTEs, window
+    functions): execute, fetch, fetchone, fetchval, transaction. you write the SQL with
+    `$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:
+    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
+    identifiers (table/column names) are interpolated, and they are quoted.
+
+errors (FAIL LOUD — unlike the mongo lib's swallow-and-default):
+    every method catches the driver error (asyncpg.PostgresError / InterfaceError, OSError
+    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
+    not wrapped, use the raw `.pool` property (the asyncpg.Pool).
+"""
+
+import logging
+from typing import Any, Dict, List, Optional, Sequence
+
+import asyncpg
+
+log = logging.getLogger(__name__)
+
+_DRIVER_ERRORS = (asyncpg.PostgresError, asyncpg.InterfaceError, OSError)
+
+
+def _quote_ident(identifier: str) -> str:
+    """quote a sql identifier (table/column), escaping embedded double-quotes
+
+    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.
+    """
+    return '"' + identifier.replace('"', '""') + '"'
+
+
+def _row_to_dict(row) -> dict:
+    """convert an asyncpg Record to a plain dict (the portable row shape)"""
+    return dict(row)
+
+
+class PsqlDB:
+    """async postgres wrapper; one pool per process, attach as app.db"""
+
+    def __init__(
+        self,
+        host: str = "localhost",
+        port: int = 5432,
+        database: Optional[str] = None,
+        user: Optional[str] = None,
+        password: Optional[str] = None,
+        *,
+        min_size: int = 1,
+        max_size: int = 10,
+        command_timeout: Optional[float] = None,
+        **pool_kwargs,
+    ):
+        """store config and build the (not-yet-connected) pool spec; no I/O here
+
+        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_
+        settings, dsn, etc). `host` may be a unix socket directory as well as a hostname.
+        """
+        self._config = dict(
+            host=host,
+            port=port,
+            database=database,
+            user=user,
+            password=password,
+            min_size=min_size,
+            max_size=max_size,
+            command_timeout=command_timeout,
+            **pool_kwargs,
+        )
+        self._pool: Optional[asyncpg.Pool] = None
+
+    async def connect(self) -> "PsqlDB":
+        """build the pool and validate it with SELECT 1; fail loud on bad config
+
+        returns self so callers can write `db = await PsqlDB(...).connect()`.
+        """
+        try:
+            self._pool = await asyncpg.create_pool(**self._config)
+            await self._pool.fetchval("SELECT 1")
+        except _DRIVER_ERRORS:
+            log.exception("psql.connect() failed")
+            raise
+        return self
+
+    async def close(self) -> None:
+        """close the pool on shutdown"""
+        if self._pool is None:
+            return
+        try:
+            await self._pool.close()
+        except _DRIVER_ERRORS:
+            log.exception("psql.close()")
+            raise
+
+    async def __aenter__(self) -> "PsqlDB":
+        return await self.connect()
+
+    async def __aexit__(self, exc_type, exc, tb) -> None:
+        await self.close()
+
+    @property
+    def pool(self) -> asyncpg.Pool:
+        """raw asyncpg.Pool escape hatch; full driver surface, raises
+
+        use for copy/prepare/listen-notify/cursors and anything not wrapped.
+        """
+        if self._pool is None:
+            raise RuntimeError("psql: not connected; call await db.connect() first")
+        return self._pool
+
+    # -------------------------------------------------------------------------
+    #   layer 2 — raw escape hatch (you write the SQL, $1 placeholders)
+
+    async def execute(self, query: str, *params: Any) -> str:
+        """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
+        the layer-1 verbs (insert/delete) which return structured values instead.
+        """
+        try:
+            return await self.pool.execute(query, *params)
+        except _DRIVER_ERRORS:
+            log.exception("psql.execute(): %s", query)
+            raise
+
+    async def fetch(self, query: str, *params: Any) -> List[dict]:
+        """run a query and return all rows as plain dicts (empty list = no rows)"""
+        try:
+            rows = await self.pool.fetch(query, *params)
+        except _DRIVER_ERRORS:
+            log.exception("psql.fetch(): %s", query)
+            raise
+        return [_row_to_dict(r) for r in rows]
+
+    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
+
+        named fetchone (not asyncpg's fetchrow) to match the mysql lib's layer-2 surface;
+        maps to the driver's fetchrow internally.
+        """
+        try:
+            row = await self.pool.fetchrow(query, *params)
+        except _DRIVER_ERRORS:
+            log.exception("psql.fetchone(): %s", query)
+            raise
+        return _row_to_dict(row) if row is not None else None
+
+    async def fetchval(self, query: str, *params: Any) -> Any:
+        """run a query and return the first column of the first row, or None"""
+        try:
+            return await self.pool.fetchval(query, *params)
+        except _DRIVER_ERRORS:
+            log.exception("psql.fetchval(): %s", query)
+            raise
+
+    def transaction(self):
+        """async context manager running a block of statements atomically
+
+        usage:
+            async with db.transaction() as conn:
+                await conn.execute("INSERT ...", a)
+                await conn.execute("UPDATE ...", b)
+        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).
+        """
+        return _Transaction(self.pool)
+
+    # -------------------------------------------------------------------------
+    #   layer 1 — friendly portable verbs (identical across psql/mysql)
+
+    async def create_database(self, name: str) -> None:
+        """CREATE DATABASE name (raises if it already exists — postgres has no IF NOT
+        EXISTS for CREATE DATABASE; catch the duplicate error or check first)"""
+        try:
+            await self.pool.execute(f"CREATE DATABASE {_quote_ident(name)}")
+        except _DRIVER_ERRORS:
+            log.exception("psql.create_database(%s)", name)
+            raise
+
+    async def create_table(self, name: str, schema: Dict[str, str]) -> None:
+        """CREATE TABLE IF NOT EXISTS from a {column: "type ..."} schema dict
+
+        schema maps column name -> its column definition, e.g.
+        {"id": "serial primary key", "name": "text not null"}. the column type/constraints
+        are caller-controlled SQL (not values), so they are interpolated as-is; only the
+        column NAME is quoted. keep the schema-dict format identical to the mysql lib.
+        """
+        cols = ", ".join(f"{_quote_ident(col)} {decl}" for col, decl in schema.items())
+        try:
+            await self.pool.execute(f"CREATE TABLE IF NOT EXISTS {_quote_ident(name)} ({cols})")
+        except _DRIVER_ERRORS:
+            log.exception("psql.create_table(%s)", name)
+            raise
+
+    async def drop(self, name: str, *, table: bool = True) -> None:
+        """DROP a table (table=True, default) or database (table=False), IF EXISTS"""
+        kind = "TABLE" if table else "DATABASE"
+        try:
+            await self.pool.execute(f"DROP {kind} IF EXISTS {_quote_ident(name)}")
+        except _DRIVER_ERRORS:
+            log.exception("psql.drop(%s)", name)
+            raise
+
+    async def insert(self, table: str, values: Dict[str, Any]) -> int:
+        """INSERT one row from {column: value}; return the inserted rowcount (1)
+
+        values are parameterized ($1, $2, ...). returns the number of rows inserted (1 on
+        success) — the portable return shape shared with the mysql lib.
+        """
+        cols = list(values.keys())
+        placeholders = ", ".join(f"${i + 1}" for i in range(len(cols)))
+        col_sql = ", ".join(_quote_ident(c) for c in cols)
+        query = f"INSERT INTO {_quote_ident(table)} ({col_sql}) VALUES ({placeholders})"
+        try:
+            status = await self.pool.execute(query, *values.values())
+        except _DRIVER_ERRORS:
+            log.exception("psql.insert(%s)", table)
+            raise
+        return _status_count(status)
+
+    async def get(self, table: str, conditions: Optional[Dict[str, Any]] = None) -> List[dict]:
+        """SELECT * rows matching equality `conditions` (col = val AND ...) as dicts
+
+        conditions=None/{} returns all rows. simple equality only — anything more complex
+        goes through raw fetch().
+        """
+        where, params = _where(conditions)
+        query = f"SELECT * FROM {_quote_ident(table)}{where}"
+        try:
+            rows = await self.pool.fetch(query, *params)
+        except _DRIVER_ERRORS:
+            log.exception("psql.get(%s)", table)
+            raise
+        return [_row_to_dict(r) for r in rows]
+
+    async def get_one(self, table: str, conditions: Optional[Dict[str, Any]] = None) -> Optional[dict]:
+        """SELECT the first row matching equality `conditions` as a dict, or None"""
+        where, params = _where(conditions)
+        query = f"SELECT * FROM {_quote_ident(table)}{where} LIMIT 1"
+        try:
+            row = await self.pool.fetchrow(query, *params)
+        except _DRIVER_ERRORS:
+            log.exception("psql.get_one(%s)", table)
+            raise
+        return _row_to_dict(row) if row is not None else None
+
+    async def delete(self, table: str, conditions: Optional[Dict[str, Any]] = None) -> int:
+        """DELETE rows matching equality `conditions`; return rowcount deleted
+
+        conditions=None/{} deletes ALL rows (documented; pass a filter to scope it).
+        """
+        where, params = _where(conditions)
+        query = f"DELETE FROM {_quote_ident(table)}{where}"
+        try:
+            status = await self.pool.execute(query, *params)
+        except _DRIVER_ERRORS:
+            log.exception("psql.delete(%s)", table)
+            raise
+        return _status_count(status)
+
+    async def exists(self, table: str, conditions: Optional[Dict[str, Any]] = None) -> bool:
+        """return whether any row matches equality `conditions`"""
+        where, params = _where(conditions)
+        query = f"SELECT EXISTS(SELECT 1 FROM {_quote_ident(table)}{where})"
+        try:
+            return bool(await self.pool.fetchval(query, *params))
+        except _DRIVER_ERRORS:
+            log.exception("psql.exists(%s)", table)
+            raise
+
+    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
+
+        `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 identical call). returns the affected rowcount.
+        """
+        cols = list(values.keys())
+        placeholders = ", ".join(f"${i + 1}" for i in range(len(cols)))
+        col_sql = ", ".join(_quote_ident(c) for c in cols)
+        conflict_sql = ", ".join(_quote_ident(c) for c in conflict)
+        updates = ", ".join(f"{_quote_ident(c)} = EXCLUDED.{_quote_ident(c)}" for c in cols)
+        query = (
+            f"INSERT INTO {_quote_ident(table)} ({col_sql}) VALUES ({placeholders}) "
+            f"ON CONFLICT ({conflict_sql}) DO UPDATE SET {updates}"
+        )
+        try:
+            status = await self.pool.execute(query, *values.values())
+        except _DRIVER_ERRORS:
+            log.exception("psql.upsert(%s)", table)
+            raise
+        return _status_count(status)
+
+
+class _Transaction:
+    """async ctx mgr: acquire a pooled connection + open a transaction on it"""
+
+    def __init__(self, pool: asyncpg.Pool):
+        self._pool = pool
+        self._conn = None
+        self._tx = None
+
+    async def __aenter__(self):
+        self._conn = await self._pool.acquire()
+        self._tx = self._conn.transaction()
+        await self._tx.start()
+        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:
+    """build a parameterized `WHERE col = $1 AND ...` clause + the params list
+
+    returns ("", []) when there are no conditions. equality only.
+    """
+    if not conditions:
+        return "", []
+    cols = list(conditions.keys())
+    clause = " AND ".join(f"{_quote_ident(c)} = ${i + 1}" for i, c in enumerate(cols))
+    return f" WHERE {clause}", list(conditions.values())
+
+
+def _status_count(status: str) -> int:
+    """parse the trailing integer out of an asyncpg status tag (e.g. 'DELETE 3' -> 3)
+
+    asyncpg returns a command tag like 'INSERT 0 1' / 'UPDATE 2' / 'DELETE 0'; the last
+    whitespace-separated token is the affected-row count. returns 0 if unparseable.
+    """
+    try:
+        return int(status.split()[-1])
+    except (ValueError, IndexError, AttributeError):
+        return 0