From f12295ce6a28665485fc55da5503a421931ee3ca Mon Sep 17 00:00:00 2001 From: disqualifier Date: Tue, 30 Jun 2026 05:48:00 -0400 Subject: [PATCH] add package: pyproject + src (layer-1 verbs + raw layer-2, fail-loud, plain-dict rows) Signed-off-by: disqualifier --- pyproject.toml | 15 ++ src/psql/__init__.py | 3 + src/psql/psql.py | 376 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 394 insertions(+) create mode 100644 pyproject.toml create mode 100644 src/psql/__init__.py create mode 100644 src/psql/psql.py diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..1de159a --- /dev/null +++ b/pyproject.toml @@ -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"] diff --git a/src/psql/__init__.py b/src/psql/__init__.py new file mode 100644 index 0000000..9dc40f3 --- /dev/null +++ b/src/psql/__init__.py @@ -0,0 +1,3 @@ +from .psql import PsqlDB + +__all__ = ["PsqlDB"] diff --git a/src/psql/psql.py b/src/psql/psql.py new file mode 100644 index 0000000..d4c4e79 --- /dev/null +++ b/src/psql/psql.py @@ -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