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>

README.md

@@ -11,24 +11,26 @@ wire-compatible and share the driver, so this covers both.
 `requirements.txt`:
 
 ```
-mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v0.1.1
+mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v1.0.0
 ```
 
 Direct:
 
 ```bash
-pip install "mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v0.1.1"
+pip install "mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v1.0.0"
 ```
 
 Pulls `aiomysql` (which pulls `PyMySQL`).
 
-Drop the `@v0.1.1` 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
 
 **Layer 1 — friendly verbs** for simple single-table CRUD. These hide the dialect and are
-**identical to the `psql` lib** — same method names, signatures, return shapes — so you
+**signature-identical to the `psql` lib** — same method names, arguments, and return
-swap psql↔mysql with zero call-site changes.
+*types* — so you swap psql↔mysql with zero call-site changes. One value-level difference:
+`insert()` returns the new row id (`lastrowid`) here vs. the inserted rowcount in psql
+(both `int`); everything else returns the same shape.
 
 **Layer 2 — raw escape hatch** for the complex ~20% (joins, aggregates, subqueries). You
 write the SQL with `%s` placeholders + a params sequence; the wrapper still gives pooling,
@@ -97,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
   `%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
-  (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
 
@@ -111,6 +118,11 @@ Layer 1 is portable — you never see these. A Layer-2 raw-SQL author does:
 `fetchone` is the unified Layer-2 name in both libs (psql maps it onto asyncpg's
 `fetchrow` internally; here it's the native DBAPI verb).
 
+`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
+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
 
 Unlike the `mongo` lib (which log-and-swallows), **this lib re-raises.** Every method
@@ -128,7 +140,10 @@ result (no row, empty table) — never a swallowed failure.
 
 `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()`
-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
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "mysql"
-version = "0.1.1"
+version = "1.0.0"
 description = "async mysql/mariadb wrapper over aiomysql: two-layer API (friendly verbs + raw escape hatch), fail-loud, config-free"
 requires-python = ">=3.10"
 dependencies = [

src/mysql/__init__.py

@@ -1,3 +1,10 @@
+from importlib.metadata import PackageNotFoundError, version
+
 from .mysql import MysqlDB
 
 __all__ = ["MysqlDB"]
+
+try:
+    __version__ = version("mysql")
+except PackageNotFoundError:
+    __version__ = "0.0.0+unknown"

src/mysql/mysql.py

@@ -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
     app.db = await MysqlDB(host="localhost", port=3306, db="app",
                            user="root", password="secret").connect()
     await app.db.insert("users", {"name": "ada", "active": True})
     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
 from typing import Any, Dict, List, Optional, Sequence
 
@@ -64,12 +33,33 @@ log = logging.getLogger(__name__)
 _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("`", "``") + "`"
 
 
@@ -92,9 +82,23 @@ 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
+        rows come back as dicts (DictCursor), overridable via pool_kwargs to a tuple
-        overridden via pool_kwargs.
+        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 "autocommit" in pool_kwargs and not pool_kwargs["autocommit"]:
+            raise ValueError(
+                f"mysql: autocommit must be on (got {pool_kwargs['autocommit']!r}) - layer-1 verbs "
+                "and raw execute() need it to persist writes (False/None/0 all disable it, and None "
+                "means 'server default' = silent data loss under autocommit=0); use db.transaction() "
+                "for atomic blocks"
+            )
         self._config = dict(
             host=host,
             port=port,
@@ -108,25 +112,54 @@ 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
 
-        returns self so callers can write `db = await MysqlDB(...).connect()`.
+        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. concurrent
+        callers serialize on an internal lock so racing connect() calls build exactly one
+        pool instead of each leaking the losers' live pools.
         """
-        try:
+        async with self._connect_lock:
-            self._pool = await aiomysql.create_pool(**self._config)
+            if self._pool is not None:
-            async with self._pool.acquire() as conn:
+                await self._close_locked()
-                async with conn.cursor() as cur:
+            pool = None
-                    await cur.execute("SELECT 1")
+            try:
-                    await cur.fetchone()
+                pool = await aiomysql.create_pool(**self._config)
-        except _DRIVER_ERRORS:
+                async with pool.acquire() as conn:
-            log.exception("mysql.connect() failed")
+                    async with conn.cursor() as cur:
-            raise
+                        await cur.execute("SELECT 1")
-        return self
+                        await cur.fetchone()
+            except _DRIVER_ERRORS:
+                log.exception("mysql.connect() failed")
+                if pool is not None:
+                    pool.close()
+                    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:
-        """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:
             return
         try:
@@ -135,11 +168,15 @@ class MysqlDB:
         except _DRIVER_ERRORS:
             log.exception("mysql.close()")
             raise
+        finally:
+            self._pool = None
 
     async def __aenter__(self) -> "MysqlDB":
+        """enter: connect() and return self"""
         return await self.connect()
 
     async def __aexit__(self, exc_type, exc, tb) -> None:
+        """exit: close(), ignoring exc_type/exc/tb"""
         await self.close()
 
     @property
@@ -155,30 +192,33 @@ class MysqlDB:
     # -------------------------------------------------------------------------
     #   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 conn.cursor() as cur:
-                await cur.execute(query, params or None)
+                await cur.execute(query, _norm_params(params))
                 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"""
         async with self.pool.acquire() as conn:
             async with conn.cursor() as cur:
-                await cur.execute(query, params or None)
+                await cur.execute(query, _norm_params(params))
                 rows = await cur.fetchall()
                 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"""
         async with self.pool.acquire() as conn:
             async with conn.cursor() as cur:
-                await cur.execute(query, params or None)
+                await cur.execute(query, _norm_params(params))
                 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:
         """run a statement (INSERT/UPDATE/DELETE/DDL); return the affected rowcount
@@ -187,7 +227,7 @@ class MysqlDB:
         lastrowid, or reach for the raw pool.
         """
         try:
-            rowcount, _ = await self._run(query, params or ())
+            rowcount, _ = await self._run(query, params)
         except _DRIVER_ERRORS:
             log.exception("mysql.execute(): %s", query)
             raise
@@ -196,7 +236,7 @@ class MysqlDB:
     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)"""
         try:
-            return await self._fetchall(query, params or ())
+            return await self._fetchall(query, params)
         except _DRIVER_ERRORS:
             log.exception("mysql.fetch(): %s", query)
             raise
@@ -204,7 +244,7 @@ class MysqlDB:
     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"""
         try:
-            return await self._fetchone(query, params or ())
+            return await self._fetchone(query, params)
         except _DRIVER_ERRORS:
             log.exception("mysql.fetchone(): %s", query)
             raise
@@ -212,34 +252,29 @@ class MysqlDB:
     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"""
         try:
-            row = await self._fetchone(query, params or ())
+            row = await self._fetchone(query, params)
         except _DRIVER_ERRORS:
             log.exception("mysql.fetchval(): %s", query)
             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
+        """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
         aiomysql connection (autocommit disabled for the block).
         """
         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:
         """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
@@ -249,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
@@ -263,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
@@ -272,7 +308,7 @@ class MysqlDB:
         """INSERT one row from {column: value}; return the new row id (lastrowid)
 
         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).
         """
         cols = list(values.keys())
@@ -289,7 +325,7 @@ class MysqlDB:
     async def get(self, table: str, conditions: Optional[Dict[str, Any]] = None) -> List[dict]:
         """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().
         """
         where, params = _where(conditions)
@@ -333,16 +369,15 @@ 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
+        """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())
         placeholders = ", ".join(["%s"] * len(cols))
@@ -376,10 +411,9 @@ class _Transaction:
         try:
             await self._conn.begin()
         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)
-            self._pool.release(self._conn)
             self._conn = None
             raise
         return self._conn
@@ -391,15 +425,26 @@ class _Transaction:
             else:
                 await self._conn.rollback()
         finally:
-            self._pool.release(self._conn)
+            # 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)
+    parts = []
-    return f" WHERE {clause}", list(conditions.values())
+    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