mysql/README.md

mysql

Async MySQL + MariaDB wrapper over aiomysql — a small, config-free, fail-loud, two-layer API: friendly portable verbs for the common case, a raw escape hatch for the rest. Third of the datastore trio (redis / psql / mysql), a sibling of the mongo lib. Class is MysqlDB. MySQL and MariaDB are wire-compatible and share the driver, so this covers both.

Install

requirements.txt:

mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v0.1.0

Direct:

pip install "mysql @ git+ssh://git@git.rethinkstudios.io/rethink-public/mysql.git@v0.1.0"

Pulls aiomysql (which pulls PyMySQL).

Drop the @v0.1.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 swap psql↔mysql with zero call-site changes.

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, parameterization, fail-loud errors, and dict rows. Raw SQL is mysql-specific.

There is deliberately nothing in between — no query builder, no ORM. A join goes through raw fetch(), never a chainable .where()/.join().

Usage

from mysql import MysqlDB

# construction is sync; connect() builds the pool + SELECT 1 to fail loud on bad config
db = await MysqlDB(host="localhost", port=3306, db="app",
                   user="root", password="secret").connect()

# --- layer 1: friendly, portable (identical to psql) ---
await db.create_table("users", {"id": "int auto_increment primary key",
                                "name": "varchar(255) not null", "age": "int"})
new_id = await db.insert("users", {"name": "ada", "age": 30})   # -> lastrowid
rows   = await db.get("users", {"age": 30})                      # [{"id": 1, "name": "ada", ...}]
one    = await db.get_one("users", {"name": "ada"})             # {...} or None
ok     = await db.exists("users", {"name": "ada"})              # True
await db.delete("users", {"name": "ada"})                       # -> rowcount
await db.upsert("users", {"id": 1, "name": "ada2"}, conflict=["id"])  # ON DUPLICATE KEY UPDATE

await db.close()

Context-manager form:

async with MysqlDB(db="app", user="root") as db:
    await db.insert("events", {"kind": "login"})

Layer 2 — raw SQL for the complex queries

# a join — not modelled by layer 1, so write it directly. %s placeholders, plain-dict rows.
rows = await db.fetch(
    "SELECT u.name, count(o.id) AS orders "
    "FROM users u JOIN orders o ON o.user_id = u.id "
    "WHERE u.active = %s GROUP BY u.name",
    (True,),
)
row  = await db.fetchone("SELECT * FROM users WHERE id = %s", (7,))   # dict or None
n    = await db.fetchval("SELECT count(*) FROM users")               # scalar or None
await db.execute("UPDATE users SET active = %s WHERE last_seen < %s", (False, cutoff))

# atomic multi-statement block — commits on clean exit, rolls back + re-raises on error
async with db.transaction() as conn:
    async with conn.cursor() as cur:
        await cur.execute("INSERT INTO ledger(acct, delta) VALUES(%s, %s)", (a, -amount))
        await cur.execute("INSERT INTO ledger(acct, delta) VALUES(%s, %s)", (b, +amount))

For anything even Layer 2 doesn't model (executemany, server-side cursors), use the raw db.pool — the underlying aiomysql.Pool.

Rows & placeholders

• Rows are plain dicts ({column: value}) via aiomysql's DictCursor — identical shape to the psql lib.
• 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.

Dialect vs psql

Layer 1 is portable — you never see these. A Layer-2 raw-SQL author does:

	mysql (this lib)	psql
placeholders	%s	$1, $2
upsert	ON DUPLICATE KEY UPDATE	ON CONFLICT ... DO UPDATE

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).

Error contract — fail loud

Unlike the mongo lib (which log-and-swallows), this lib re-raises. Every method catches the driver error (pymysql.err.MySQLError, OSError on connection loss), logs it via logging.getLogger(__name__), and raises. A None / [] return is only ever a real result (no row, empty table) — never a swallowed failure.

Surface

• Layer 1: create_database, create_table(name, {col: "type"}), drop(name, table=), insert(table, {col: val}), get(table, {conds}), get_one(table, {conds}), delete(table, {conds}), exists(table, {conds}), upsert(table, {col: val}, conflict=[...])
• Layer 2: execute(sql, params=None), fetch, fetchone, fetchval, transaction()
• Raw: pool property → aiomysql.Pool

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.

Versioning

Releases are tagged vX.Y.Z. The install line above pins a release; drop the @vX.Y.Z suffix to install the latest unpinned. Pin deliberately for reproducible installs.