diff --git a/README.md b/README.md index 70da3ac..9ae951f 100644 --- a/README.md +++ b/README.md @@ -11,18 +11,18 @@ 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.5 +mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v0.1.6 ``` Direct: ```bash -pip install "mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v0.1.5" +pip install "mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v0.1.6" ``` Pulls `aiomysql` (which pulls `PyMySQL`). -Drop the `@v0.1.5` suffix from the line above to install the latest unpinned. +Drop the `@v0.1.6` suffix from the line above to install the latest unpinned. ## The two-layer API diff --git a/pyproject.toml b/pyproject.toml index 45148e3..681a8b0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "mysql" -version = "0.1.5" +version = "0.1.6" description = "async mysql/mariadb wrapper over aiomysql: two-layer API (friendly verbs + raw escape hatch), fail-loud, config-free" requires-python = ">=3.10" dependencies = [ diff --git a/src/mysql/mysql.py b/src/mysql/mysql.py index 1434b45..58092e5 100644 --- a/src/mysql/mysql.py +++ b/src/mysql/mysql.py @@ -104,23 +104,40 @@ class MysqlDB: """ async with self._connect_lock: if self._pool is not None: - await self.close() - pool = await aiomysql.create_pool(**self._config) + await self._close_locked() + pool = None try: + pool = await aiomysql.create_pool(**self._config) async with pool.acquire() as conn: async with conn.cursor() as cur: await cur.execute("SELECT 1") 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: - log.exception("mysql.connect() validation failed") - pool.close() - await pool.wait_closed() + 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: @@ -151,26 +168,32 @@ class MysqlDB: # ------------------------------------------------------------------------- # internal cursor helpers (autocommit pool) - async def _run(self, query: str, params: Sequence = ()): - """execute a statement on a pooled cursor; return (rowcount, lastrowid)""" + 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 + 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 or None) + await cur.execute(query, params if params is None else tuple(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, params if params is None else tuple(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, params if params is None else tuple(params)) return await cur.fetchone() # ------------------------------------------------------------------------- @@ -183,7 +206,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 @@ -192,7 +215,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 @@ -200,7 +223,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 @@ -208,7 +231,7 @@ 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 @@ -230,7 +253,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(name)}", ()) except _DRIVER_ERRORS: log.exception("mysql.create_database(%s)", name) raise @@ -245,7 +268,7 @@ class MysqlDB: """ cols = ", ".join(f"{_quote_ident(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(name)} ({cols})", ()) except _DRIVER_ERRORS: log.exception("mysql.create_table(%s)", name) raise @@ -254,7 +277,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(name)}", ()) except _DRIVER_ERRORS: log.exception("mysql.drop(%s)", name) raise