3 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
3 changed files with 27 additions and 41 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.1.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 = [
+23 -37
View File
@@ -26,6 +26,7 @@ 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
@@ -198,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)
@@ -217,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()
# ------------------------------------------------------------------------- # -------------------------------------------------------------------------
@@ -230,7 +233,7 @@ 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.debug("mysql.execute(): %s", query, exc_info=True) log.debug("mysql.execute(): %s", query, exc_info=True)
raise raise
@@ -239,7 +242,7 @@ class MysqlDB:
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.debug("mysql.fetch(): %s", query, exc_info=True) log.debug("mysql.fetch(): %s", query, exc_info=True)
raise raise
@@ -247,7 +250,7 @@ class MysqlDB:
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.debug("mysql.fetchone(): %s", query, exc_info=True) log.debug("mysql.fetchone(): %s", query, exc_info=True)
raise raise
@@ -255,7 +258,7 @@ class MysqlDB:
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.debug("mysql.fetchval(): %s", query, exc_info=True) log.debug("mysql.fetchval(): %s", query, exc_info=True)
raise raise
@@ -269,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)
@@ -398,39 +401,22 @@ class MysqlDB:
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: