add package: pyproject + src (layer-1 identical to psql, raw layer-2, fail-loud, dict rows)

Signed-off-by: disqualifier <dev@disqualifier.me>

pyproject.toml

@@ -0,0 +1,15 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "mysql"
+version = "0.1.0"
+description = "async mysql/mariadb wrapper over aiomysql: two-layer API (friendly verbs + raw escape hatch), fail-loud, config-free"
+requires-python = ">=3.10"
+dependencies = [
+    "aiomysql>=0.2",
+]
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/mysql"]

src/mysql/__init__.py

@@ -0,0 +1,3 @@
+from .mysql import MysqlDB
+
+__all__ = ["MysqlDB"]

src/mysql/mysql.py

@@ -0,0 +1,397 @@
+"""
+async mysql / mariadb wrapper over aiomysql — two-layer API (friendly verbs + raw hatch)
+
+covers MySQL and MariaDB (wire-compatible, same driver).
+
+object pattern (one pool per process), attach to the app:
+    from mysql import MysqlDB
+    app.db = await MysqlDB(host="localhost", port=3306, db="app",
+                           user="root", 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 MysqlDB(db="app", user="root") as db:
+        await db.execute("CREATE TABLE ...")
+
+lifecycle:
+    construction is sync and opens no socket. connect() builds the aiomysql 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 and waits for it.
+    the pool runs with autocommit=True, so the layer-1 verbs and raw execute() commit on
+    their own; transaction() opens an explicit transaction for atomic multi-statement work.
+
+two-layer API (IDENTICAL layer-1 surface to the `psql` lib — swap psql<->mysql with zero
+call-site changes; only the dialect below differs, hidden inside the wrapper):
+    LAYER 1 — friendly portable verbs for simple single-table CRUD: create_database,
+    create_table, drop, insert, get, get_one, delete, exists, upsert.
+    LAYER 2 — raw escape hatch for the complex ~20% (joins, aggregates, subqueries): you
+    write the SQL with `%s` placeholders + a params sequence; the wrapper gives pooling,
+    parameterization, fail-loud errors, and dict rows. execute, fetch, fetchone, fetchval,
+    transaction. NOTHING in between — no query builder / ORM.
+
+dialect (mysql-specific; visible only in raw layer-2 SQL you write):
+    - placeholders are `%s` (here) vs `$1` (psql). the `%s` is the DBAPI placeholder the
+      driver consumes, NOT python `%` string formatting.
+    - upsert emits `ON DUPLICATE KEY UPDATE` (here) vs `ON CONFLICT` (psql); layer-1
+      upsert() hides this.
+
+rows:
+    layer 1 and fetch/fetchone return plain dicts ({column: value}) via aiomysql's
+    DictCursor — identical shape to the psql lib.
+
+placeholders / injection safety:
+    values are ALWAYS parameterized — layer 1 builds `%s` internally; layer 2 takes your
+    `%s` placeholders + a params sequence. never f-string/%-format a value into SQL. only
+    identifiers (table/column names) are interpolated, and they are backtick-quoted.
+
+errors (FAIL LOUD — unlike the mongo lib's swallow-and-default):
+    every method catches the driver error (pymysql.err.MySQLError, 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 aiomysql.Pool).
+"""
+
+import logging
+from typing import Any, Dict, List, Optional, Sequence
+
+import aiomysql
+from pymysql.err import MySQLError
+
+log = logging.getLogger(__name__)
+
+_DRIVER_ERRORS = (MySQLError, OSError)
+
+
+def _quote_ident(identifier: str) -> str:
+    """backtick-quote a sql identifier (table/column), escaping embedded backticks
+
+    identifiers can't be parameterized, so they are interpolated — backtick-quoting +
+    doubling any embedded backtick is the mysql-safe way to do that for caller names.
+    """
+    return "`" + identifier.replace("`", "``") + "`"
+
+
+class MysqlDB:
+    """async mysql/mariadb wrapper; one pool per process, attach as app.db"""
+
+    def __init__(
+        self,
+        host: str = "localhost",
+        port: int = 3306,
+        db: Optional[str] = None,
+        user: Optional[str] = None,
+        password: str = "",
+        *,
+        minsize: int = 1,
+        maxsize: int = 10,
+        **pool_kwargs,
+    ):
+        """store config and build the (not-yet-connected) pool spec; no I/O here
+
+        host/port/db/user/password/minsize/maxsize are injected by the caller. extra
+        pool_kwargs pass through to aiomysql.create_pool (unix_socket, ssl, charset, ...).
+        rows come back as dicts (DictCursor) and autocommit is on by default; both can be
+        overridden via pool_kwargs.
+        """
+        self._config = dict(
+            host=host,
+            port=port,
+            db=db,
+            user=user,
+            password=password,
+            minsize=minsize,
+            maxsize=maxsize,
+        )
+        self._config.setdefault("autocommit", True)
+        self._config.setdefault("cursorclass", aiomysql.cursors.DictCursor)
+        self._config.update(pool_kwargs)
+        self._pool: Optional[aiomysql.Pool] = None
+
+    async def connect(self) -> "MysqlDB":
+        """build the pool and validate it with SELECT 1; fail loud on bad config
+
+        returns self so callers can write `db = await MysqlDB(...).connect()`.
+        """
+        try:
+            self._pool = await aiomysql.create_pool(**self._config)
+            async with self._pool.acquire() as conn:
+                async with conn.cursor() as cur:
+                    await cur.execute("SELECT 1")
+                    await cur.fetchone()
+        except _DRIVER_ERRORS:
+            log.exception("mysql.connect() failed")
+            raise
+        return self
+
+    async def close(self) -> None:
+        """close the pool and wait for it on shutdown"""
+        if self._pool is None:
+            return
+        try:
+            self._pool.close()
+            await self._pool.wait_closed()
+        except _DRIVER_ERRORS:
+            log.exception("mysql.close()")
+            raise
+
+    async def __aenter__(self) -> "MysqlDB":
+        return await self.connect()
+
+    async def __aexit__(self, exc_type, exc, tb) -> None:
+        await self.close()
+
+    @property
+    def pool(self) -> aiomysql.Pool:
+        """raw aiomysql.Pool escape hatch; full driver surface, raises
+
+        use for executemany/server-side cursors and anything not wrapped.
+        """
+        if self._pool is None:
+            raise RuntimeError("mysql: not connected; call await db.connect() first")
+        return self._pool
+
+    # -------------------------------------------------------------------------
+    #   internal cursor helpers (autocommit pool)
+
+    async def _run(self, query: str, params: Sequence = ()):
+        """execute a statement on a pooled cursor; return (rowcount, lastrowid)"""
+        async with self.pool.acquire() as conn:
+            async with conn.cursor() as cur:
+                await cur.execute(query, params or None)
+                return cur.rowcount, cur.lastrowid
+
+    async def _fetchall(self, query: str, params: Sequence = ()) -> List[dict]:
+        """execute a query and return all rows as dicts"""
+        async with self.pool.acquire() as conn:
+            async with conn.cursor() as cur:
+                await cur.execute(query, params or None)
+                rows = await cur.fetchall()
+                return list(rows)
+
+    async def _fetchone(self, query: str, params: Sequence = ()) -> Optional[dict]:
+        """execute a query and return the first row dict, or None"""
+        async with self.pool.acquire() as conn:
+            async with conn.cursor() as cur:
+                await cur.execute(query, params or None)
+                return await cur.fetchone()
+
+    # -------------------------------------------------------------------------
+    #   layer 2 — raw escape hatch (you write the SQL, %s placeholders)
+
+    async def execute(self, query: str, params: Optional[Sequence] = None) -> int:
+        """run a statement (INSERT/UPDATE/DELETE/DDL); return the affected rowcount
+
+        for an INSERT where you need the new id, use insert() (layer 1) which returns
+        lastrowid, or reach for the raw pool.
+        """
+        try:
+            rowcount, _ = await self._run(query, params or ())
+        except _DRIVER_ERRORS:
+            log.exception("mysql.execute(): %s", query)
+            raise
+        return rowcount
+
+    async def fetch(self, query: str, params: Optional[Sequence] = None) -> List[dict]:
+        """run a query and return all rows as plain dicts (empty list = no rows)"""
+        try:
+            return await self._fetchall(query, params or ())
+        except _DRIVER_ERRORS:
+            log.exception("mysql.fetch(): %s", query)
+            raise
+
+    async def fetchone(self, query: str, params: Optional[Sequence] = None) -> Optional[dict]:
+        """run a query and return the first row as a dict, or None if no rows"""
+        try:
+            return await self._fetchone(query, params or ())
+        except _DRIVER_ERRORS:
+            log.exception("mysql.fetchone(): %s", query)
+            raise
+
+    async def fetchval(self, query: str, params: Optional[Sequence] = None) -> Any:
+        """run a query and return the first column of the first row, or None"""
+        try:
+            row = await self._fetchone(query, params or ())
+        except _DRIVER_ERRORS:
+            log.exception("mysql.fetchval(): %s", query)
+            raise
+        if not row:
+            return None
+        return next(iter(row.values()))
+
+    def transaction(self):
+        """async context manager running a block of statements atomically
+
+        usage:
+            async with db.transaction() as conn:
+                async with conn.cursor() as cur:
+                    await cur.execute("INSERT ...", (a,))
+                    await cur.execute("UPDATE ...", (b,))
+        commits on clean exit, rolls back and re-raises on any error. `conn` is a raw
+        aiomysql connection (autocommit disabled for the block).
+        """
+        return _Transaction(self.pool)
+
+    # -------------------------------------------------------------------------
+    #   layer 1 — friendly portable verbs (identical to the psql lib)
+
+    async def create_database(self, name: str) -> None:
+        """CREATE DATABASE IF NOT EXISTS name"""
+        try:
+            await self._run(f"CREATE DATABASE IF NOT EXISTS {_quote_ident(name)}")
+        except _DRIVER_ERRORS:
+            log.exception("mysql.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": "int auto_increment primary key", "name": "varchar(255) not null"}. the
+        column type/constraints are caller-controlled SQL (not values), interpolated
+        as-is; only the column NAME is quoted. same schema-dict format as the psql lib.
+        """
+        cols = ", ".join(f"{_quote_ident(col)} {decl}" for col, decl in schema.items())
+        try:
+            await self._run(f"CREATE TABLE IF NOT EXISTS {_quote_ident(name)} ({cols})")
+        except _DRIVER_ERRORS:
+            log.exception("mysql.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._run(f"DROP {kind} IF EXISTS {_quote_ident(name)}")
+        except _DRIVER_ERRORS:
+            log.exception("mysql.drop(%s)", name)
+            raise
+
+    async def insert(self, table: str, values: Dict[str, Any]) -> int:
+        """INSERT one row from {column: value}; return the new row id (lastrowid)
+
+        values are parameterized (%s). returns lastrowid for an auto-increment table, or
+        0 when there is no auto-increment key — the portable return shape shared with the
+        psql lib (which returns the RETURNING rowcount).
+        """
+        cols = list(values.keys())
+        placeholders = ", ".join(["%s"] * len(cols))
+        col_sql = ", ".join(_quote_ident(c) for c in cols)
+        query = f"INSERT INTO {_quote_ident(table)} ({col_sql}) VALUES ({placeholders})"
+        try:
+            _, lastrowid = await self._run(query, list(values.values()))
+        except _DRIVER_ERRORS:
+            log.exception("mysql.insert(%s)", table)
+            raise
+        return lastrowid
+
+    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:
+            return await self._fetchall(query, params)
+        except _DRIVER_ERRORS:
+            log.exception("mysql.get(%s)", table)
+            raise
+
+    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:
+            return await self._fetchone(query, params)
+        except _DRIVER_ERRORS:
+            log.exception("mysql.get_one(%s)", table)
+            raise
+
+    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:
+            rowcount, _ = await self._run(query, params)
+        except _DRIVER_ERRORS:
+            log.exception("mysql.delete(%s)", table)
+            raise
+        return rowcount
+
+    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:
+            row = await self._fetchone(query, params)
+        except _DRIVER_ERRORS:
+            log.exception("mysql.exists(%s)", table)
+            raise
+        return bool(next(iter(row.values()))) if row else False
+
+    async def upsert(self, table: str, values: Dict[str, Any], conflict: Sequence[str]) -> int:
+        """INSERT ... ON DUPLICATE KEY UPDATE — insert or update on unique/pk clash
+
+        `conflict` is the list of columns forming the unique/pk constraint. mysql upserts
+        on ANY unique-key clash (it does not name the constraint the way postgres does), so
+        `conflict` is accepted for a call-signature identical to psql but the constraint is
+        whatever unique/pk key the row violates. the wrapper emits ON DUPLICATE KEY UPDATE
+        (psql emits ON CONFLICT for the identical call). returns the affected rowcount.
+        """
+        cols = list(values.keys())
+        placeholders = ", ".join(["%s"] * len(cols))
+        col_sql = ", ".join(_quote_ident(c) for c in cols)
+        updates = ", ".join(f"{_quote_ident(c)} = VALUES({_quote_ident(c)})" for c in cols)
+        query = (
+            f"INSERT INTO {_quote_ident(table)} ({col_sql}) VALUES ({placeholders}) "
+            f"ON DUPLICATE KEY UPDATE {updates}"
+        )
+        try:
+            rowcount, _ = await self._run(query, list(values.values()))
+        except _DRIVER_ERRORS:
+            log.exception("mysql.upsert(%s)", table)
+            raise
+        return rowcount
+
+
+class _Transaction:
+    """async ctx mgr: acquire a pooled connection + run a transaction on it
+
+    autocommit is disabled for the block so the statements commit together on clean
+    exit (or roll back on error).
+    """
+
+    def __init__(self, pool: aiomysql.Pool):
+        self._pool = pool
+        self._conn = None
+
+    async def __aenter__(self):
+        self._conn = await self._pool.acquire()
+        await self._conn.begin()
+        return self._conn
+
+    async def __aexit__(self, exc_type, exc, tb) -> None:
+        try:
+            if exc_type is None:
+                await self._conn.commit()
+            else:
+                await self._conn.rollback()
+        finally:
+            self._pool.release(self._conn)
+
+
+def _where(conditions: Optional[Dict[str, Any]]) -> tuple:
+    """build a parameterized `WHERE col = %s AND ...` clause + the params list
+
+    returns ("", []) when there are no conditions. equality only.
+    """
+    if not conditions:
+        return "", []
+    clause = " AND ".join(f"{_quote_ident(c)} = %s" for c in conditions)
+    return f" WHERE {clause}", list(conditions.values())