5 Commits
Author SHA1 Message Date
dsql 3ee0b0a39e chore: bump to 1.2.0
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-10-16 04:32:47 -04:00
dsql dc9ab8a45c refactor: transaction() as an asynccontextmanager over aiomysql's acquire context
T3-21: the _Transaction class hand-rolled acquire / start / commit-or-rollback /
release plus a leak guard for a failed start. an asynccontextmanager over the
driver's own pool.acquire() context does the same, and the acquire context
releases on every path including a failed start. verified on a real backend:
commit, rollback + re-raise, cancellation rollback, failed-start leak guard
(pool max 1) and release after 50 transactions.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-10-16 04:32:47 -04:00
dsql 7eb5d552f5 fix: layer-1 verbs always substitute params, so a % identifier reaches the server intact
T3-16: _norm_params turned an empty params list into None, so aiomysql skipped
query % args and a no-condition get/get_one/exists/delete on a table named
50%_off sent `50%%_off` (table not found). layer 1 now passes its params
list as-is; only the layer-2 methods normalize.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-10-16 04:21:29 -04:00
dsql febace7d5d chore: bump to 1.1.0 (logging-discipline audit)
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-08-10 23:00:50 -04:00
dsql 6eb3c2f841 fix: log-XOR-raise — demote the pre-raise log to DEBUG, revise the contract (D1)
every wrapped method did log.exception (ERROR + traceback) immediately before re-raising
the driver error — the same failure reported at two layers (the log AND the raised
exception). that is the log-and-raise violation: a method that re-raises must not also
error-log, because the caller — which alone knows whether the failure is fatal or routine
— is the one that logs. demote all wrapped-method logs to log.debug(..., exc_info=True):
the traceback stays available at DEBUG, and the raised exception is the single loud
terminal signal. docstring updated from "logs via getLogger and re-raises" to the
corrected raise-XOR-log contract. no behavior change beyond log level — the fail-loud
re-raise is unchanged.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-08-09 02:18:08 -04:00
3 changed files with 47 additions and 58 deletions

No files matched your search

+3 -3
View File
@@ -11,18 +11,18 @@ wire-compatible and share the driver, so this covers both.
`requirements.txt`: `requirements.txt`:
``` ```
mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v1.0.0 mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v1.2.0
``` ```
Direct: Direct:
```bash ```bash
pip install "mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v1.0.0" pip install "mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v1.2.0"
``` ```
Pulls `aiomysql` (which pulls `PyMySQL`). Pulls `aiomysql` (which pulls `PyMySQL`).
Drop the `@v1.0.0` suffix from the line above to install the latest unpinned. Drop the `@v1.2.0` suffix from the line above to install the latest unpinned.
## The two-layer API ## The two-layer API
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "mysql" name = "mysql"
version = "1.0.0" version = "1.2.0"
description = "async mysql/mariadb wrapper over aiomysql: two-layer API (friendly verbs + raw escape hatch), fail-loud, config-free" description = "async mysql/mariadb wrapper over aiomysql: two-layer API (friendly verbs + raw escape hatch), fail-loud, config-free"
requires-python = ">=3.10" requires-python = ">=3.10"
dependencies = [ dependencies = [
+43 -54
View File
@@ -17,12 +17,16 @@ a literal `%` in layer-2 SQL text must be written `%%` only when params are pass
1 handles this for you. 1 handles this for you.
errors are FAIL LOUD (unlike the mongo lib's swallow-and-default): every method catches errors are FAIL LOUD (unlike the mongo lib's swallow-and-default): every method catches
the driver error, logs via getLogger(__name__), and re-raises. a None/[] return is only the driver error and re-raises it - the raised exception IS the signal, and the caller
ever a real result, never a swallowed failure. for anything not wrapped, use `.pool`. (which alone knows whether a failure is fatal or routine) decides and logs. the wrapped
method itself logs only at DEBUG (with the traceback) so a failure isn't reported twice
(raise XOR log). a None/[] return is only ever a real result, never a swallowed failure.
for anything not wrapped, use `.pool`.
""" """
import asyncio import asyncio
import logging import logging
from contextlib import asynccontextmanager
from typing import Any, Dict, List, Optional, Sequence from typing import Any, Dict, List, Optional, Sequence
import aiomysql import aiomysql
@@ -135,7 +139,7 @@ class MysqlDB:
await cur.execute("SELECT 1") await cur.execute("SELECT 1")
await cur.fetchone() await cur.fetchone()
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.connect() failed") log.debug("mysql.connect() failed", exc_info=True)
if pool is not None: if pool is not None:
pool.close() pool.close()
await pool.wait_closed() await pool.wait_closed()
@@ -166,7 +170,7 @@ class MysqlDB:
self._pool.close() self._pool.close()
await self._pool.wait_closed() await self._pool.wait_closed()
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.close()") log.debug("mysql.close()", exc_info=True)
raise raise
finally: finally:
self._pool = None self._pool = None
@@ -195,18 +199,20 @@ class MysqlDB:
async def _run(self, query: str, params: Optional[Sequence] = None): async def _run(self, query: str, params: Optional[Sequence] = None):
"""execute a statement on a pooled cursor; return (rowcount, lastrowid) """execute a statement on a pooled cursor; return (rowcount, lastrowid)
params go through `_norm_params` (see it for the no-substitution/str-bytes handling) params are passed to aiomysql as-is: layer 1 always hands a list (even empty) so
`query % args` runs and collapses `_quote_ident`'s `%%`; layer 2 normalizes first
via `_norm_params`
""" """
async with self.pool.acquire() as conn: async with self.pool.acquire() as conn:
async with conn.cursor() as cur: async with conn.cursor() as cur:
await cur.execute(query, _norm_params(params)) await cur.execute(query, params)
return cur.rowcount, cur.lastrowid return cur.rowcount, cur.lastrowid
async def _fetchall(self, query: str, params: Optional[Sequence] = None) -> List[dict]: async def _fetchall(self, query: str, params: Optional[Sequence] = None) -> List[dict]:
"""execute a query and return all rows as dicts""" """execute a query and return all rows as dicts"""
async with self.pool.acquire() as conn: async with self.pool.acquire() as conn:
async with conn.cursor() as cur: async with conn.cursor() as cur:
await cur.execute(query, _norm_params(params)) await cur.execute(query, params)
rows = await cur.fetchall() rows = await cur.fetchall()
return list(rows) return list(rows)
@@ -214,7 +220,7 @@ class MysqlDB:
"""execute a query and return the first row dict, or None""" """execute a query and return the first row dict, or None"""
async with self.pool.acquire() as conn: async with self.pool.acquire() as conn:
async with conn.cursor() as cur: async with conn.cursor() as cur:
await cur.execute(query, _norm_params(params)) await cur.execute(query, params)
return await cur.fetchone() return await cur.fetchone()
# ------------------------------------------------------------------------- # -------------------------------------------------------------------------
@@ -227,34 +233,34 @@ class MysqlDB:
lastrowid, or reach for the raw pool. lastrowid, or reach for the raw pool.
""" """
try: try:
rowcount, _ = await self._run(query, params) rowcount, _ = await self._run(query, _norm_params(params))
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.execute(): %s", query) log.debug("mysql.execute(): %s", query, exc_info=True)
raise raise
return rowcount return rowcount
async def fetch(self, query: str, params: Optional[Sequence] = None) -> List[dict]: 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)""" """run a query and return all rows as plain dicts (empty list = no rows)"""
try: try:
return await self._fetchall(query, params) return await self._fetchall(query, _norm_params(params))
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.fetch(): %s", query) log.debug("mysql.fetch(): %s", query, exc_info=True)
raise raise
async def fetchone(self, query: str, params: Optional[Sequence] = None) -> Optional[dict]: 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""" """run a query and return the first row as a dict, or None if no rows"""
try: try:
return await self._fetchone(query, params) return await self._fetchone(query, _norm_params(params))
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.fetchone(): %s", query) log.debug("mysql.fetchone(): %s", query, exc_info=True)
raise raise
async def fetchval(self, query: str, params: Optional[Sequence] = None) -> Any: 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""" """run a query and return the first column of the first row, or None"""
try: try:
row = await self._fetchone(query, params) row = await self._fetchone(query, _norm_params(params))
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.fetchval(): %s", query) log.debug("mysql.fetchval(): %s", query, exc_info=True)
raise raise
if not row: if not row:
return None return None
@@ -266,7 +272,7 @@ class MysqlDB:
commits on clean exit, rolls back and re-raises on any error. `conn` is a raw commits on clean exit, rolls back and re-raises on any error. `conn` is a raw
aiomysql connection (autocommit disabled for the block). aiomysql connection (autocommit disabled for the block).
""" """
return _Transaction(self.pool) return _transaction(self.pool)
# ------------------------------------------------------------------------- # -------------------------------------------------------------------------
# layer 1 - friendly portable verbs (identical to the psql lib) # layer 1 - friendly portable verbs (identical to the psql lib)
@@ -276,7 +282,7 @@ class MysqlDB:
try: try:
await self._run(f"CREATE DATABASE IF NOT EXISTS {_quote_ident_literal(name)}") await self._run(f"CREATE DATABASE IF NOT EXISTS {_quote_ident_literal(name)}")
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.create_database(%s)", name) log.debug("mysql.create_database(%s)", name, exc_info=True)
raise raise
async def create_table(self, name: str, schema: Dict[str, str]) -> None: async def create_table(self, name: str, schema: Dict[str, str]) -> None:
@@ -292,7 +298,7 @@ class MysqlDB:
try: try:
await self._run(f"CREATE TABLE IF NOT EXISTS {_quote_ident_literal(name)} ({cols})") await self._run(f"CREATE TABLE IF NOT EXISTS {_quote_ident_literal(name)} ({cols})")
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.create_table(%s)", name) log.debug("mysql.create_table(%s)", name, exc_info=True)
raise raise
async def drop(self, name: str, *, table: bool = True) -> None: async def drop(self, name: str, *, table: bool = True) -> None:
@@ -301,7 +307,7 @@ class MysqlDB:
try: try:
await self._run(f"DROP {kind} IF EXISTS {_quote_ident_literal(name)}") await self._run(f"DROP {kind} IF EXISTS {_quote_ident_literal(name)}")
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.drop(%s)", name) log.debug("mysql.drop(%s)", name, exc_info=True)
raise raise
async def insert(self, table: str, values: Dict[str, Any]) -> int: async def insert(self, table: str, values: Dict[str, Any]) -> int:
@@ -318,7 +324,7 @@ class MysqlDB:
try: try:
_, lastrowid = await self._run(query, list(values.values())) _, lastrowid = await self._run(query, list(values.values()))
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.insert(%s)", table) log.debug("mysql.insert(%s)", table, exc_info=True)
raise raise
return lastrowid return lastrowid
@@ -333,7 +339,7 @@ class MysqlDB:
try: try:
return await self._fetchall(query, params) return await self._fetchall(query, params)
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.get(%s)", table) log.debug("mysql.get(%s)", table, exc_info=True)
raise raise
async def get_one(self, table: str, conditions: Optional[Dict[str, Any]] = None) -> Optional[dict]: async def get_one(self, table: str, conditions: Optional[Dict[str, Any]] = None) -> Optional[dict]:
@@ -343,7 +349,7 @@ class MysqlDB:
try: try:
return await self._fetchone(query, params) return await self._fetchone(query, params)
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.get_one(%s)", table) log.debug("mysql.get_one(%s)", table, exc_info=True)
raise raise
async def delete(self, table: str, conditions: Optional[Dict[str, Any]] = None) -> int: async def delete(self, table: str, conditions: Optional[Dict[str, Any]] = None) -> int:
@@ -356,7 +362,7 @@ class MysqlDB:
try: try:
rowcount, _ = await self._run(query, params) rowcount, _ = await self._run(query, params)
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.delete(%s)", table) log.debug("mysql.delete(%s)", table, exc_info=True)
raise raise
return rowcount return rowcount
@@ -367,7 +373,7 @@ class MysqlDB:
try: try:
row = await self._fetchone(query, params) row = await self._fetchone(query, params)
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.exists(%s)", table) log.debug("mysql.exists(%s)", table, exc_info=True)
raise raise
return bool(_first_value(row)) if row else False return bool(_first_value(row)) if row else False
@@ -390,44 +396,27 @@ class MysqlDB:
try: try:
rowcount, _ = await self._run(query, list(values.values())) rowcount, _ = await self._run(query, list(values.values()))
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.upsert(%s)", table) log.debug("mysql.upsert(%s)", table, exc_info=True)
raise raise
return rowcount return rowcount
class _Transaction: @asynccontextmanager
"""async ctx mgr: acquire a pooled connection + run a transaction on it async def _transaction(pool: aiomysql.Pool):
"""acquire a pooled connection and run the block in a transaction on it
autocommit is disabled for the block so the statements commit together on clean begin() disables autocommit for the block; commit on clean exit, rollback + re-raise
exit (or roll back on error). on any error. the acquire context releases the connection on every path - including
a failed begin(), so a failed start never burns a pool slot
""" """
async with pool.acquire() as conn:
def __init__(self, pool: aiomysql.Pool): await conn.begin()
self._pool = pool
self._conn = None
async def __aenter__(self):
self._conn = await self._pool.acquire()
try: try:
await self._conn.begin() yield conn
except BaseException: except BaseException:
# __aexit__ never runs if __aenter__ raises, so begin() failing after acquire await conn.rollback()
# would otherwise leak the pooled connection - release it here instead.
await self._pool.release(self._conn)
self._conn = None
raise raise
return self._conn await conn.commit()
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:
# release() returns a task (schedules a _wakeup); await it to avoid leaving
# it un-awaited each transaction, matching psql's and aiomysql's own idiom.
await self._pool.release(self._conn)
def _where(conditions: Optional[Dict[str, Any]]) -> tuple: def _where(conditions: Optional[Dict[str, Any]]) -> tuple: