rethink-public/mysql
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
dsql
2026-06-30 20:12:40 -04:00
parent 9beeee4d9f
commit 9034ce32c2
3 changed files with 415 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+15
View File
@@ -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
+3
View File
@@ -0,0 +1,3 @@
from .mysql import MysqlDB
__all__ = ["MysqlDB"]
src/mysql/mysql.py
+397
View File
@@ -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())