7 Commits
Author SHA1 Message Date
dsql 2055ad978a release: 1.0.0
first stable release. pre-1.0.0 verification complete: all surviving MED regressions and
gaps resolved and independently re-fired, tree audited clean across the suite.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-09 18:53:15 -04:00
dsql 1ffb45dd9d fix: no query%args on no-param DDL (literal % survives); don't shred a str/bytes param; reject autocommit=None
three mysql-only defects from the fable deep-scan, all from the 0e01c4e/50388de waves:
- create_table/create_database/drop passed an explicit () that forced query%args, so a
  literal % in a caller's column decl (COMMENT 'save 10%') raised a TypeError that bypassed
  the _DRIVER_ERRORS wrapper. no-param statements now pass None (no substitution) and quote
  identifiers via _quote_ident_literal (no %-doubling, since nothing collapses it).
- _run/_fetchall/_fetchone blind-tuple(params) exploded a bare str/bytes param into
  per-character args; _norm_params now passes a str/bytes through as one bound value and
  normalizes an empty sequence to None. multi-param tuples/lists/dicts unchanged.
- the autocommit guard checked 'is False' only, so autocommit=None slipped past and, against
  a server with autocommit=0, silently never committed; it now rejects any non-True value.
psql is unaffected (asyncpg binds $1 with no query%args) - Layer-1 signatures still byte-match.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-06 16:38:05 -04:00
dsql eba5cd41f6 docs: add __aenter__/__aexit__ one-liner docstrings for twin parity with redis
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-06 02:55:57 -04:00
dsql 0e01c4e77f fix: %-escape identifiers only on the params path (no more %% wrong-target on DDL)
_quote_ident() always doubles % to %%, but aiomysql only runs query % args
substitution when args is not None - _run/_fetchall/_fetchone were passing
params or None, so no-params DDL (create_database/create_table/drop) and
no-condition get/delete/exists shipped literal %% to the server, silently
targeting the wrong object and making IF EXISTS no-op (regression 2e837da).

_run/_fetchall/_fetchone now forward None as-is (preserving layer-2's
documented "no params, no substitution" contract) but always pass a real
tuple otherwise, so layer-1's %%-escaped identifiers always collapse back
to a single % as intended. The three no-params DDL verbs now pass an
explicit empty tuple so substitution runs for them too.

connect()/close() had the same lock asymmetry as psql (close() wasn't
guarded by _connect_lock, so a close() racing an in-flight connect() could
no-op while the new pool went live) - fixed in lockstep with the psql twin
fix, and create_pool() is now inside the same try/log/re-raise as the
SELECT-1 validation.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-03 19:11:39 -04:00
dsql ebf1f80652 refactor: derive __version__ from package metadata (single source)
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-03 17:00:32 -04:00
dsql f45db389ba docs: compress prose/module docstrings, em-dash->hyphen (de-bloat wave 1)
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-03 00:16:12 -04:00
dsql 2e837da7df fix: close() nulls _pool + connect() lock (psql-7/8 twin), % in identifiers, doc caveats
close() now nulls self._pool so a closed instance reports not-connected instead of
masquerading as live (twin of psql-7); connect() is guarded by an internal asyncio.Lock
so concurrent connect() calls serialize instead of racing to create and orphan multiple
live pools (twin of psql-8). _quote_ident now escapes a literal % in identifiers (mysql-7:
PyMySQL's query % args substitution otherwise breaks any Layer-1 call against a %-bearing
table/column name). fetchval/exists tolerate a caller-overridden tuple cursorclass via a
new _first_value helper (mysql-8). Documents the %%-escaping rule for literal % in raw SQL
text (mysql-6) and the upsert() VALUES() deprecation on MySQL 8.0.20+ (mysql-9).

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-02 23:32:53 -04:00
4 changed files with 154 additions and 122 deletions
+15 -6
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@v0.1.3 mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v1.0.0
``` ```
Direct: Direct:
```bash ```bash
pip install "mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v0.1.3" pip install "mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v1.0.0"
``` ```
Pulls `aiomysql` (which pulls `PyMySQL`). Pulls `aiomysql` (which pulls `PyMySQL`).
Drop the `@v0.1.3` suffix from the line above to install the latest unpinned. Drop the `@v1.0.0` suffix from the line above to install the latest unpinned.
## The two-layer API ## The two-layer API
@@ -99,7 +99,12 @@ For anything even Layer 2 doesn't model (`executemany`, server-side cursors), us
- **Values are always parameterized.** Layer 1 builds `%s` internally; Layer 2 takes your - **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 (the `%s` placeholders + a params sequence. Never f-string/`%`-format a value into SQL (the
`%s` is the DBAPI placeholder, not Python string formatting). Only identifiers `%s` is the DBAPI placeholder, not Python string formatting). Only identifiers
(table/column names) are interpolated, and they're backtick-quoted. (table/column names) are interpolated, and they're backtick-quoted — Layer 1 also
escapes a literal `%` inside an identifier for you.
- **A literal `%` in Layer-2 SQL text must be written `%%`.** PyMySQL builds the query via
`query % args` whenever you pass params, so `"... LIKE '%foo%'"` with params raises
`TypeError`/`ValueError` — write `"... LIKE '%%foo%%'"` instead. Calls with no params are
unaffected (no substitution happens), and Layer 1 handles this for you internally.
## Dialect vs psql ## Dialect vs psql
@@ -115,7 +120,8 @@ Layer 1 is portable — you never see these. A Layer-2 raw-SQL author does:
`upsert()`'s returned rowcount follows MySQL's `ON DUPLICATE KEY UPDATE` convention (1 for `upsert()`'s returned rowcount follows MySQL's `ON DUPLICATE KEY UPDATE` convention (1 for
an insert, 2 for an update, 0 for a no-op) — a per-row count that differs from psql's; treat an insert, 2 for an update, 0 for a no-op) — a per-row count that differs from psql's; treat
it as "affected", not "rows matched". it as "affected", not "rows matched". It emits the `col = VALUES(col)` update form, which
MySQL 8.0.20+ deprecates (warning 1287, still functional; MariaDB is unaffected).
## Error contract — fail loud ## Error contract — fail loud
@@ -134,7 +140,10 @@ result (no row, empty table) — never a swallowed failure.
`get`/`delete`/`exists`/`get_one` take **equality** conditions only (`col = val AND ...`); `get`/`delete`/`exists`/`get_one` take **equality** conditions only (`col = val AND ...`);
anything richer goes through Layer 2. The pool runs with `autocommit=True`; `transaction()` anything richer goes through Layer 2. The pool runs with `autocommit=True`; `transaction()`
opens an explicit transaction for atomic multi-statement blocks. opens an explicit transaction for atomic multi-statement blocks. `cursorclass` is
overridable via a constructor kwarg (default `DictCursor`); `fetchval`/`exists` work under
either a dict or tuple cursorclass, everything else stays dict-shaped only under the
default `DictCursor`.
## Versioning ## Versioning
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "mysql" name = "mysql"
version = "0.1.3" version = "1.0.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 = [
+7
View File
@@ -1,3 +1,10 @@
from importlib.metadata import PackageNotFoundError, version
from .mysql import MysqlDB from .mysql import MysqlDB
__all__ = ["MysqlDB"] __all__ = ["MysqlDB"]
try:
__version__ = version("mysql")
except PackageNotFoundError:
__version__ = "0.0.0+unknown"
+131 -115
View File
@@ -1,58 +1,27 @@
""" """
async mysql / mariadb wrapper over aiomysql two-layer API (friendly verbs + raw hatch) 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 from mysql import MysqlDB
app.db = await MysqlDB(host="localhost", port=3306, db="app", app.db = await MysqlDB(host="localhost", port=3306, db="app",
user="root", password="secret").connect() user="root", password="secret").connect()
await app.db.insert("users", {"name": "ada", "active": True}) await app.db.insert("users", {"name": "ada", "active": True})
rows = await app.db.get("users", {"active": True}) # [{"name": "ada", ...}] rows = await app.db.get("users", {"active": True}) # [{"name": "ada", ...}]
await app.db.close() # on shutdown await app.db.close()
context manager: layer 1 is IDENTICAL to the `psql` lib's surface (swap psql<->mysql with zero call-site
async with MysqlDB(db="app", user="root") as db: changes); layer 2 is the raw `%s`-placeholder escape hatch. see README for the full API
await db.execute("CREATE TABLE ...") and mysql/psql dialect differences.
lifecycle: a literal `%` in layer-2 SQL text must be written `%%` only when params are passed
construction is sync and opens no socket. connect() builds the aiomysql pool and (aiomysql runs `query % args` only then); with no params the text is sent verbatim. layer
validates it with `SELECT 1` so a bad host/credentials fails loud immediately rather 1 handles this for you.
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 errors are FAIL LOUD (unlike the mongo lib's swallow-and-default): every method catches
call-site changes; only the dialect below differs, hidden inside the wrapper): the driver error, logs via getLogger(__name__), and re-raises. a None/[] return is only
LAYER 1 — friendly portable verbs for simple single-table CRUD: create_database, ever a real result, never a swallowed failure. for anything not wrapped, use `.pool`.
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 asyncio
import logging import logging
from typing import Any, Dict, List, Optional, Sequence from typing import Any, Dict, List, Optional, Sequence
@@ -64,12 +33,33 @@ log = logging.getLogger(__name__)
_DRIVER_ERRORS = (MySQLError, OSError) _DRIVER_ERRORS = (MySQLError, OSError)
def _quote_ident(identifier: str) -> str: def _first_value(row) -> Any:
"""backtick-quote a sql identifier (table/column), escaping embedded backticks """return the first column's value from a dict or sequence row (either cursorclass)"""
return next(iter(row.values())) if isinstance(row, dict) else row[0]
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. def _norm_params(params: Optional[Sequence]) -> Any:
""" """normalize params for aiomysql: None/empty-sequence -> None (no `query % args`, so a
literal `%` is sent as-is); a str/bytes stays a single bound value (never exploded per
char); anything else passes through"""
if params is None:
return None
if isinstance(params, (str, bytes, bytearray)):
return params
if isinstance(params, Sequence) and len(params) == 0:
return None
return params
def _quote_ident(identifier: str) -> str:
"""backtick-quote an identifier for the params path, doubling `` ` `` and `%` (the `%%`
collapses back under `query % args`); use `_quote_ident_literal` on the no-params path"""
return "`" + identifier.replace("`", "``").replace("%", "%%") + "`"
def _quote_ident_literal(identifier: str) -> str:
"""backtick-quote an identifier for the no-params DDL path; no `%`-doubling (no
substitution runs to collapse it), only the backtick escaped"""
return "`" + identifier.replace("`", "``") + "`" return "`" + identifier.replace("`", "``") + "`"
@@ -92,16 +82,22 @@ class MysqlDB:
host/port/db/user/password/minsize/maxsize are injected by the caller. extra 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, ...). pool_kwargs pass through to aiomysql.create_pool (unix_socket, ssl, charset, ...).
rows come back as dicts (DictCursor), overridable via pool_kwargs. autocommit is rows come back as dicts (DictCursor), overridable via pool_kwargs to a tuple
always on: layer-1 verbs and raw execute() rely on it to commit on their own, so cursorclass - fetchval/exists work either way, everything else stays dict-shaped
`autocommit=False` in pool_kwargs is rejected (use transaction() for atomic only under DictCursor. autocommit is always on: layer-1 verbs and raw execute()
multi-statement blocks instead — it disables autocommit for just that block). rely on it to commit their own writes without an explicit commit, so
`autocommit=False` in pool_kwargs is rejected outright - silently accepting it
would mean every write sits uncommitted until something else commits or the
connection is returned to the pool and rolled back, losing data with no error.
use transaction() for an atomic multi-statement block instead (it disables
autocommit for just that block).
""" """
if pool_kwargs.get("autocommit") is False: if "autocommit" in pool_kwargs and not pool_kwargs["autocommit"]:
raise ValueError( raise ValueError(
"mysql: autocommit=False is not supported via pool_kwargs — layer-1 verbs and " f"mysql: autocommit must be on (got {pool_kwargs['autocommit']!r}) - layer-1 verbs "
"raw execute() need autocommit to persist their writes; use db.transaction() " "and raw execute() need it to persist writes (False/None/0 all disable it, and None "
"for an atomic multi-statement block instead" "means 'server default' = silent data loss under autocommit=0); use db.transaction() "
"for atomic blocks"
) )
self._config = dict( self._config = dict(
host=host, host=host,
@@ -116,6 +112,7 @@ class MysqlDB:
self._config.setdefault("cursorclass", aiomysql.cursors.DictCursor) self._config.setdefault("cursorclass", aiomysql.cursors.DictCursor)
self._config.update(pool_kwargs) self._config.update(pool_kwargs)
self._pool: Optional[aiomysql.Pool] = None self._pool: Optional[aiomysql.Pool] = None
self._connect_lock = asyncio.Lock()
async def connect(self) -> "MysqlDB": async def connect(self) -> "MysqlDB":
"""build the pool and validate it with SELECT 1; fail loud on bad config """build the pool and validate it with SELECT 1; fail loud on bad config
@@ -123,26 +120,46 @@ class MysqlDB:
returns self so callers can write `db = await MysqlDB(...).connect()`. if called returns self so callers can write `db = await MysqlDB(...).connect()`. if called
again on an already-connected instance the previous pool is closed first (no again on an already-connected instance the previous pool is closed first (no
orphaned pool); if the SELECT-1 validation fails the freshly-built pool is torn orphaned pool); if the SELECT-1 validation fails the freshly-built pool is torn
down before re-raising, so a failed connect() never leaks a live pool. down before re-raising, so a failed connect() never leaks a live pool. concurrent
callers serialize on an internal lock so racing connect() calls build exactly one
pool instead of each leaking the losers' live pools.
""" """
if self._pool is not None: async with self._connect_lock:
await self.close() if self._pool is not None:
pool = await aiomysql.create_pool(**self._config) await self._close_locked()
try: pool = None
async with pool.acquire() as conn: try:
async with conn.cursor() as cur: pool = await aiomysql.create_pool(**self._config)
await cur.execute("SELECT 1") async with pool.acquire() as conn:
await cur.fetchone() async with conn.cursor() as cur:
except BaseException: await cur.execute("SELECT 1")
log.exception("mysql.connect() validation failed") await cur.fetchone()
pool.close() except _DRIVER_ERRORS:
await pool.wait_closed() log.exception("mysql.connect() failed")
raise if pool is not None:
self._pool = pool pool.close()
return self await pool.wait_closed()
raise
except BaseException:
if pool is not None:
pool.close()
await pool.wait_closed()
raise
self._pool = pool
return self
async def close(self) -> None: async def close(self) -> None:
"""close the pool and wait for it on shutdown""" """close the pool and wait for it on shutdown
guarded by the same lock as connect() - a close() racing an in-flight connect()
waits for it rather than no-opping against a not-yet-installed pool and leaving
the just-built one live.
"""
async with self._connect_lock:
await self._close_locked()
async def _close_locked(self) -> None:
"""close the pool and wait for it; caller must hold `_connect_lock`"""
if self._pool is None: if self._pool is None:
return return
try: try:
@@ -151,11 +168,15 @@ class MysqlDB:
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.close()") log.exception("mysql.close()")
raise raise
finally:
self._pool = None
async def __aenter__(self) -> "MysqlDB": async def __aenter__(self) -> "MysqlDB":
"""enter: connect() and return self"""
return await self.connect() return await self.connect()
async def __aexit__(self, exc_type, exc, tb) -> None: async def __aexit__(self, exc_type, exc, tb) -> None:
"""exit: close(), ignoring exc_type/exc/tb"""
await self.close() await self.close()
@property @property
@@ -171,30 +192,33 @@ class MysqlDB:
# ------------------------------------------------------------------------- # -------------------------------------------------------------------------
# internal cursor helpers (autocommit pool) # internal cursor helpers (autocommit pool)
async def _run(self, query: str, params: Sequence = ()): 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)
"""
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, params or None) await cur.execute(query, _norm_params(params))
return cur.rowcount, cur.lastrowid return cur.rowcount, cur.lastrowid
async def _fetchall(self, query: str, params: Sequence = ()) -> 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, params or None) await cur.execute(query, _norm_params(params))
rows = await cur.fetchall() rows = await cur.fetchall()
return list(rows) return list(rows)
async def _fetchone(self, query: str, params: Sequence = ()) -> Optional[dict]: async def _fetchone(self, query: str, params: Optional[Sequence] = None) -> Optional[dict]:
"""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, params or None) await cur.execute(query, _norm_params(params))
return await cur.fetchone() return await cur.fetchone()
# ------------------------------------------------------------------------- # -------------------------------------------------------------------------
# layer 2 raw escape hatch (you write the SQL, %s placeholders) # layer 2 - raw escape hatch (you write the SQL, %s placeholders)
async def execute(self, query: str, params: Optional[Sequence] = None) -> int: async def execute(self, query: str, params: Optional[Sequence] = None) -> int:
"""run a statement (INSERT/UPDATE/DELETE/DDL); return the affected rowcount """run a statement (INSERT/UPDATE/DELETE/DDL); return the affected rowcount
@@ -203,7 +227,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 or ()) rowcount, _ = await self._run(query, params)
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.execute(): %s", query) log.exception("mysql.execute(): %s", query)
raise raise
@@ -212,7 +236,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 or ()) return await self._fetchall(query, params)
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.fetch(): %s", query) log.exception("mysql.fetch(): %s", query)
raise raise
@@ -220,7 +244,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 or ()) return await self._fetchone(query, params)
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.fetchone(): %s", query) log.exception("mysql.fetchone(): %s", query)
raise raise
@@ -228,34 +252,29 @@ 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 or ()) row = await self._fetchone(query, params)
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.fetchval(): %s", query) log.exception("mysql.fetchval(): %s", query)
raise raise
if not row: if not row:
return None return None
return next(iter(row.values())) return _first_value(row)
def transaction(self): def transaction(self):
"""async context manager running a block of statements atomically """async context manager running a block of statements atomically (see README)
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 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)
async def create_database(self, name: str) -> None: async def create_database(self, name: str) -> None:
"""CREATE DATABASE IF NOT EXISTS name""" """CREATE DATABASE IF NOT EXISTS name"""
try: try:
await self._run(f"CREATE DATABASE IF NOT EXISTS {_quote_ident(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.exception("mysql.create_database(%s)", name)
raise raise
@@ -265,12 +284,13 @@ class MysqlDB:
schema maps column name -> its column definition, e.g. schema maps column name -> its column definition, e.g.
{"id": "int auto_increment primary key", "name": "varchar(255) not null"}. the {"id": "int auto_increment primary key", "name": "varchar(255) not null"}. the
column type/constraints are caller-controlled SQL (not values), interpolated column type/constraints are caller-controlled SQL interpolated as-is (a literal `%`
as-is; only the column NAME is quoted. same schema-dict format as the psql lib. in a decl is sent unaltered - no params); 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()) cols = ", ".join(f"{_quote_ident_literal(col)} {decl}" for col, decl in schema.items())
try: try:
await self._run(f"CREATE TABLE IF NOT EXISTS {_quote_ident(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.exception("mysql.create_table(%s)", name)
raise raise
@@ -279,7 +299,7 @@ class MysqlDB:
"""DROP a table (table=True, default) or database (table=False), IF EXISTS""" """DROP a table (table=True, default) or database (table=False), IF EXISTS"""
kind = "TABLE" if table else "DATABASE" kind = "TABLE" if table else "DATABASE"
try: try:
await self._run(f"DROP {kind} IF EXISTS {_quote_ident(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.exception("mysql.drop(%s)", name)
raise raise
@@ -288,7 +308,7 @@ class MysqlDB:
"""INSERT one row from {column: value}; return the new row id (lastrowid) """INSERT one row from {column: value}; return the new row id (lastrowid)
values are parameterized (%s). returns lastrowid for an auto-increment table, or 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 0 when there is no auto-increment key - the portable return shape shared with the
psql lib (which returns the RETURNING rowcount). psql lib (which returns the RETURNING rowcount).
""" """
cols = list(values.keys()) cols = list(values.keys())
@@ -305,7 +325,7 @@ class MysqlDB:
async def get(self, table: str, conditions: Optional[Dict[str, Any]] = None) -> List[dict]: async def get(self, table: str, conditions: Optional[Dict[str, Any]] = None) -> List[dict]:
"""SELECT * rows matching equality `conditions` (col = val AND ...) as dicts """SELECT * rows matching equality `conditions` (col = val AND ...) as dicts
conditions=None/{} returns all rows. simple equality only anything more complex conditions=None/{} returns all rows. simple equality only - anything more complex
goes through raw fetch(). goes through raw fetch().
""" """
where, params = _where(conditions) where, params = _where(conditions)
@@ -349,16 +369,15 @@ class MysqlDB:
except _DRIVER_ERRORS: except _DRIVER_ERRORS:
log.exception("mysql.exists(%s)", table) log.exception("mysql.exists(%s)", table)
raise raise
return bool(next(iter(row.values()))) if row else False return bool(_first_value(row)) if row else False
async def upsert(self, table: str, values: Dict[str, Any], conflict: Sequence[str]) -> int: 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 """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 `conflict` is accepted for call-signature parity with psql, but mysql upserts on
on ANY unique-key clash (it does not name the constraint the way postgres does), so ANY unique/pk clash (it does not name the constraint like postgres). returns the
`conflict` is accepted for a call-signature identical to psql but the constraint is affected rowcount. emits the `col = VALUES(col)` update form, deprecated since
whatever unique/pk key the row violates. the wrapper emits ON DUPLICATE KEY UPDATE MySQL 8.0.20 (warning 1287, still functional); unaffected on MariaDB.
(psql emits ON CONFLICT for the identical call). returns the affected rowcount.
""" """
cols = list(values.keys()) cols = list(values.keys())
placeholders = ", ".join(["%s"] * len(cols)) placeholders = ", ".join(["%s"] * len(cols))
@@ -392,9 +411,8 @@ class _Transaction:
try: try:
await self._conn.begin() await self._conn.begin()
except BaseException: except BaseException:
# begin() failing after acquire would otherwise leak the pooled connection — # __aexit__ never runs if __aenter__ raises, so begin() failing after acquire
# __aexit__ is not called when __aenter__ raises. release it and reset so a # would otherwise leak the pooled connection - release it here instead.
# failed transaction start never burns a pool slot.
await self._pool.release(self._conn) await self._pool.release(self._conn)
self._conn = None self._conn = None
raise raise
@@ -407,10 +425,8 @@ class _Transaction:
else: else:
await self._conn.rollback() await self._conn.rollback()
finally: finally:
# aiomysql's release() returns a task (it schedules a _wakeup); await it so we # release() returns a task (schedules a _wakeup); await it to avoid leaving
# don't leave an un-awaited task each transaction matches psql's awaited # it un-awaited each transaction, matching psql's and aiomysql's own idiom.
# release and aiomysql's own idiom. the connection is returned synchronously
# inside release() regardless, so pool accounting is never at risk.
await self._pool.release(self._conn) await self._pool.release(self._conn)