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>

src/mysql/mysql.py

@@ -12,8 +12,9 @@ layer 1 is IDENTICAL to the `psql` lib's surface (swap psql<->mysql with zero ca
 changes); layer 2 is the raw `%s`-placeholder escape hatch. see README for the full API
 and mysql/psql dialect differences.
 
-a literal `%` in layer-2 SQL text must be written `%%` (PyMySQL builds the query via
+a literal `%` in layer-2 SQL text must be written `%%` only when params are passed
-`query % args` whenever params are passed); layer 1 escapes `%` in identifiers for you.
+(aiomysql runs `query % args` only then); with no params the text is sent verbatim. layer
+1 handles this for you.
 
 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
@@ -37,11 +38,31 @@ def _first_value(row) -> Any:
     return next(iter(row.values())) if isinstance(row, dict) else row[0]
 
 
+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 a sql identifier, doubling embedded backticks and `%` (see module `%%` note)"""
+    """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("`", "``") + "`"
+
+
 class MysqlDB:
     """async mysql/mariadb wrapper; one pool per process, attach as app.db"""
 
@@ -71,11 +92,12 @@ class MysqlDB:
         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(
-                "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(
             host=host,
@@ -173,21 +195,18 @@ class MysqlDB:
     async def _run(self, query: str, params: Optional[Sequence] = None):
         """execute a statement on a pooled cursor; return (rowcount, lastrowid)
 
-        `params=None` is forwarded as-is (aiomysql skips `query % args` entirely - the
+        params go through `_norm_params` (see it for the no-substitution/str-bytes handling)
-        layer-2 "no params, no substitution" contract); any other value, including an
-        empty sequence, is passed as a real tuple so `query % args` always runs and
-        collapses layer-1's `%%`-escaped identifiers back to `%` (see module note).
         """
         async with self.pool.acquire() as conn:
             async with conn.cursor() as cur:
-                await cur.execute(query, params if params is None else tuple(params))
+                await cur.execute(query, _norm_params(params))
                 return cur.rowcount, cur.lastrowid
 
     async def _fetchall(self, query: str, params: Optional[Sequence] = None) -> 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 if params is None else tuple(params))
+                await cur.execute(query, _norm_params(params))
                 rows = await cur.fetchall()
                 return list(rows)
 
@@ -195,7 +214,7 @@ class MysqlDB:
         """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 if params is None else tuple(params))
+                await cur.execute(query, _norm_params(params))
                 return await cur.fetchone()
 
     # -------------------------------------------------------------------------
@@ -255,7 +274,7 @@ class MysqlDB:
     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)}", ())
+            await self._run(f"CREATE DATABASE IF NOT EXISTS {_quote_ident_literal(name)}")
         except _DRIVER_ERRORS:
             log.exception("mysql.create_database(%s)", name)
             raise
@@ -265,12 +284,13 @@ class MysqlDB:
 
         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
+        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:
-            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:
             log.exception("mysql.create_table(%s)", name)
             raise
@@ -279,7 +299,7 @@ class MysqlDB:
         """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)}", ())
+            await self._run(f"DROP {kind} IF EXISTS {_quote_ident_literal(name)}")
         except _DRIVER_ERRORS:
             log.exception("mysql.drop(%s)", name)
             raise