diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..451d2b1 --- /dev/null +++ b/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"] diff --git a/src/mysql/__init__.py b/src/mysql/__init__.py new file mode 100644 index 0000000..79935b7 --- /dev/null +++ b/src/mysql/__init__.py @@ -0,0 +1,3 @@ +from .mysql import MysqlDB + +__all__ = ["MysqlDB"] diff --git a/src/mysql/mysql.py b/src/mysql/mysql.py new file mode 100644 index 0000000..c6241ec --- /dev/null +++ b/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())