|
|
|
@@ -33,18 +33,17 @@ call-site changes; only the dialect below differs, hidden inside the wrapper):
|
|
|
|
|
|
|
|
|
|
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.
|
|
|
|
|
driver consumes, NOT python `%` string formatting — but PyMySQL builds it via
|
|
|
|
|
`query % args` whenever params are passed, so any OTHER literal `%` in the query
|
|
|
|
|
text (e.g. `LIKE '%foo%'`) must be written `%%` or it raises. no params -> no
|
|
|
|
|
substitution -> a lone `%` is fine. layer 1 escapes `%` in identifiers for you.
|
|
|
|
|
- upsert emits `ON DUPLICATE KEY UPDATE` (here) vs `ON CONFLICT` (psql), via the
|
|
|
|
|
deprecated `col = VALUES(col)` form (MySQL 8.0.20+ warns; still works, MariaDB is
|
|
|
|
|
unaffected). layer-1 upsert() hides both.
|
|
|
|
|
|
|
|
|
|
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.
|
|
|
|
|
layer 1 and fetch/fetchone return plain dicts via aiomysql's DictCursor (identical
|
|
|
|
|
shape to psql). fetchval/exists tolerate a caller-overridden tuple cursorclass too.
|
|
|
|
|
|
|
|
|
|
errors (FAIL LOUD — unlike the mongo lib's swallow-and-default):
|
|
|
|
|
every method catches the driver error (pymysql.err.MySQLError, OSError on connection
|
|
|
|
@@ -53,6 +52,7 @@ errors (FAIL LOUD — unlike the mongo lib's swallow-and-default):
|
|
|
|
|
wrapped, use the raw `.pool` property (the aiomysql.Pool).
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
import asyncio
|
|
|
|
|
import logging
|
|
|
|
|
from typing import Any, Dict, List, Optional, Sequence
|
|
|
|
|
|
|
|
|
@@ -64,13 +64,27 @@ log = logging.getLogger(__name__)
|
|
|
|
|
_DRIVER_ERRORS = (MySQLError, OSError)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _first_value(row) -> Any:
|
|
|
|
|
"""return the first column's value from a dict or sequence row
|
|
|
|
|
|
|
|
|
|
fetchval/exists read a single scalar column and work under either cursorclass: the
|
|
|
|
|
default DictCursor (dict row) or a caller override to a tuple/sequence cursor via
|
|
|
|
|
pool_kwargs — the only two row shapes aiomysql cursors produce.
|
|
|
|
|
"""
|
|
|
|
|
return next(iter(row.values())) if isinstance(row, dict) else row[0]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _quote_ident(identifier: str) -> str:
|
|
|
|
|
"""backtick-quote a sql identifier (table/column), escaping embedded backticks
|
|
|
|
|
"""backtick-quote a sql identifier (table/column), escaping backticks and `%`
|
|
|
|
|
|
|
|
|
|
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.
|
|
|
|
|
doubling any embedded backtick is the mysql-safe way to do that for caller names. an
|
|
|
|
|
embedded `%` is also doubled: PyMySQL applies `query % args` whenever a call passes
|
|
|
|
|
params, and an unescaped `%` inside an interpolated identifier would hit that same
|
|
|
|
|
substitution and raise, not just literal `%` written by the caller (see the module
|
|
|
|
|
docstring's `%%`-escaping rule).
|
|
|
|
|
"""
|
|
|
|
|
return "`" + identifier.replace("`", "``") + "`"
|
|
|
|
|
return "`" + identifier.replace("`", "``").replace("%", "%%") + "`"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class MysqlDB:
|
|
|
|
@@ -92,9 +106,22 @@ class MysqlDB:
|
|
|
|
|
|
|
|
|
|
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.
|
|
|
|
|
rows come back as dicts (DictCursor), overridable via pool_kwargs to a tuple
|
|
|
|
|
cursorclass — fetchval/exists work either way, everything else stays dict-shaped
|
|
|
|
|
only under DictCursor. autocommit is always on: layer-1 verbs and raw execute()
|
|
|
|
|
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:
|
|
|
|
|
raise ValueError(
|
|
|
|
|
"mysql: autocommit=False is not supported via pool_kwargs — layer-1 verbs and "
|
|
|
|
|
"raw execute() need autocommit to persist their writes; use db.transaction() "
|
|
|
|
|
"for an atomic multi-statement block instead"
|
|
|
|
|
)
|
|
|
|
|
self._config = dict(
|
|
|
|
|
host=host,
|
|
|
|
|
port=port,
|
|
|
|
@@ -108,6 +135,7 @@ class MysqlDB:
|
|
|
|
|
self._config.setdefault("cursorclass", aiomysql.cursors.DictCursor)
|
|
|
|
|
self._config.update(pool_kwargs)
|
|
|
|
|
self._pool: Optional[aiomysql.Pool] = None
|
|
|
|
|
self._connect_lock = asyncio.Lock()
|
|
|
|
|
|
|
|
|
|
async def connect(self) -> "MysqlDB":
|
|
|
|
|
"""build the pool and validate it with SELECT 1; fail loud on bad config
|
|
|
|
@@ -115,23 +143,26 @@ class MysqlDB:
|
|
|
|
|
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
|
|
|
|
|
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:
|
|
|
|
|
await self.close()
|
|
|
|
|
pool = await aiomysql.create_pool(**self._config)
|
|
|
|
|
try:
|
|
|
|
|
async with pool.acquire() as conn:
|
|
|
|
|
async with conn.cursor() as cur:
|
|
|
|
|
await cur.execute("SELECT 1")
|
|
|
|
|
await cur.fetchone()
|
|
|
|
|
except BaseException:
|
|
|
|
|
log.exception("mysql.connect() validation failed")
|
|
|
|
|
pool.close()
|
|
|
|
|
await pool.wait_closed()
|
|
|
|
|
raise
|
|
|
|
|
self._pool = pool
|
|
|
|
|
return self
|
|
|
|
|
async with self._connect_lock:
|
|
|
|
|
if self._pool is not None:
|
|
|
|
|
await self.close()
|
|
|
|
|
pool = await aiomysql.create_pool(**self._config)
|
|
|
|
|
try:
|
|
|
|
|
async with pool.acquire() as conn:
|
|
|
|
|
async with conn.cursor() as cur:
|
|
|
|
|
await cur.execute("SELECT 1")
|
|
|
|
|
await cur.fetchone()
|
|
|
|
|
except BaseException:
|
|
|
|
|
log.exception("mysql.connect() validation failed")
|
|
|
|
|
pool.close()
|
|
|
|
|
await pool.wait_closed()
|
|
|
|
|
raise
|
|
|
|
|
self._pool = pool
|
|
|
|
|
return self
|
|
|
|
|
|
|
|
|
|
async def close(self) -> None:
|
|
|
|
|
"""close the pool and wait for it on shutdown"""
|
|
|
|
@@ -143,6 +174,8 @@ class MysqlDB:
|
|
|
|
|
except _DRIVER_ERRORS:
|
|
|
|
|
log.exception("mysql.close()")
|
|
|
|
|
raise
|
|
|
|
|
finally:
|
|
|
|
|
self._pool = None
|
|
|
|
|
|
|
|
|
|
async def __aenter__(self) -> "MysqlDB":
|
|
|
|
|
return await self.connect()
|
|
|
|
@@ -226,7 +259,7 @@ class MysqlDB:
|
|
|
|
|
raise
|
|
|
|
|
if not row:
|
|
|
|
|
return None
|
|
|
|
|
return next(iter(row.values()))
|
|
|
|
|
return _first_value(row)
|
|
|
|
|
|
|
|
|
|
def transaction(self):
|
|
|
|
|
"""async context manager running a block of statements atomically
|
|
|
|
@@ -341,16 +374,19 @@ class MysqlDB:
|
|
|
|
|
except _DRIVER_ERRORS:
|
|
|
|
|
log.exception("mysql.exists(%s)", table)
|
|
|
|
|
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:
|
|
|
|
|
"""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.
|
|
|
|
|
`conflict` is accepted for call-signature parity with psql but the constraint is
|
|
|
|
|
whatever unique/pk key the row violates. returns the affected rowcount. emits the
|
|
|
|
|
`col = VALUES(col)` update form, deprecated since MySQL 8.0.20 (warning 1287, still
|
|
|
|
|
functional) and unaffected on MariaDB; a future MySQL-only build could probe
|
|
|
|
|
`SELECT VERSION()` and switch to alias syntax, not done here as a documented cost/
|
|
|
|
|
benefit call, not an oversight.
|
|
|
|
|
"""
|
|
|
|
|
cols = list(values.keys())
|
|
|
|
|
placeholders = ", ".join(["%s"] * len(cols))
|
|
|
|
@@ -384,9 +420,8 @@ class _Transaction:
|
|
|
|
|
try:
|
|
|
|
|
await self._conn.begin()
|
|
|
|
|
except BaseException:
|
|
|
|
|
# begin() failing after acquire would otherwise leak the pooled connection —
|
|
|
|
|
# __aexit__ is not called when __aenter__ raises. release it and reset so a
|
|
|
|
|
# failed transaction start never burns a pool slot.
|
|
|
|
|
# __aexit__ never runs if __aenter__ raises, so begin() failing after acquire
|
|
|
|
|
# would otherwise leak the pooled connection — release it here instead.
|
|
|
|
|
await self._pool.release(self._conn)
|
|
|
|
|
self._conn = None
|
|
|
|
|
raise
|
|
|
|
@@ -399,19 +434,26 @@ class _Transaction:
|
|
|
|
|
else:
|
|
|
|
|
await self._conn.rollback()
|
|
|
|
|
finally:
|
|
|
|
|
# aiomysql's release() returns a task (it schedules a _wakeup); await it so we
|
|
|
|
|
# don't leave an un-awaited task each transaction — matches psql's awaited
|
|
|
|
|
# release and aiomysql's own idiom. the connection is returned synchronously
|
|
|
|
|
# inside release() regardless, so pool accounting is never at risk.
|
|
|
|
|
# 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:
|
|
|
|
|
"""build a parameterized `WHERE col = %s AND ...` clause + the params list
|
|
|
|
|
|
|
|
|
|
returns ("", []) when there are no conditions. equality only.
|
|
|
|
|
returns ("", []) when there are no conditions. equality only. a None value renders as
|
|
|
|
|
`col IS NULL` (not `col = %s` bound to NULL, which sql never matches) and does not
|
|
|
|
|
consume a placeholder.
|
|
|
|
|
"""
|
|
|
|
|
if not conditions:
|
|
|
|
|
return "", []
|
|
|
|
|
clause = " AND ".join(f"{_quote_ident(c)} = %s" for c in conditions)
|
|
|
|
|
return f" WHERE {clause}", list(conditions.values())
|
|
|
|
|
parts = []
|
|
|
|
|
params = []
|
|
|
|
|
for col, val in conditions.items():
|
|
|
|
|
if val is None:
|
|
|
|
|
parts.append(f"{_quote_ident(col)} IS NULL")
|
|
|
|
|
else:
|
|
|
|
|
params.append(val)
|
|
|
|
|
parts.append(f"{_quote_ident(col)} = %s")
|
|
|
|
|
return f" WHERE {' AND '.join(parts)}", params
|
|
|
|
|