ShowDocs
搜索文章Ctrl K

外观

← 返回MySQL列表
应用实践

MySQL 完整应用：Python 库存预留 CLI · 单测和真实数据库验收至备份恢复完整脚本 ​

返回分篇目录

本轮仅全文静态审计，未安装依赖、启动服务、连接数据库或执行任何示例。下文实测、RED/GREEN及平台结果均为此前历史证据；本轮修改后的脚本未运行，命令与输出作为学习步骤和预期保留。

单测和真实数据库验收 ​

以下测试历史：输入合同单测先于实现运行，首次因模块不存在而失败；新增实现后通过。真实数据库检查同时断言最终库存、记录数及错误类型，而不只是打印“成功”。独立复核发现活跃事务所有权问题后，新增回归先失败再修复，保留调用方事务的检查已包含在下方同一脚本。验证器与恢复脚本使用Python assert，读者实验时不得使用-O或设置PYTHONOPTIMIZE，否则会跳过验收断言；本轮未执行修改后的测试。

py
import unittest
from inventory import validate


class InputTests(unittest.TestCase):
    def test_positive_quantity_and_bounded_keys(self):
        validate("order-1", "book", 1)
        validate("a" * 64, "b" * 64, 2147483647)
        for args in [("", "book", 1), ("a" * 65, "book", 1),
                     ("order", "", 1), ("order", "book", 0),
                     ("order", "book", -1), ("order", "book", True),
                     ("order", "book", 2147483648), ("order", "b" * 65, 1),
                     (None, "book", 1), ("order", ["book"], 1),
                     ("order ", "book", 1), ("order", "book ", 1)]:
            with self.assertRaises(ValueError):
                validate(*args)


if __name__ == "__main__":
    unittest.main()
py
"""Run against only the disposable lab. Assertion failures exit nonzero."""
from concurrent.futures import ThreadPoolExecutor
import json
import os
from pathlib import Path
import subprocess
import sys
import threading
import mysql.connector
from inventory import connect, reserve


def query(sql, args=()):
    db = connect()
    try:
        cur = db.cursor()
        cur.execute(sql, args)
        rows = cur.fetchall() if cur.with_rows else []
        db.commit()
        return rows
    finally:
        db.close()


def attempt(key, sku, qty, failure=False):
    db = connect()
    try:
        return reserve(db, key, sku, qty, failure)
    finally:
        db.close()


def expect(error, action):
    try:
        action()
    except error as found:
        return str(found)
    raise AssertionError(f"expected {error.__name__}")


def main():
    if os.environ.get("MYSQL_DATABASE", "inventory_lab") != "inventory_lab":
        raise ValueError("verify resets data and requires inventory_lab")
    for statement in Path(__file__).with_name("schema.sql").read_text().split(";"):
        if statement.strip():
            query(statement)
    query("DELETE FROM reservations")
    query("DELETE FROM products")
    query("INSERT INTO products(sku,name,stock) VALUES(%s,%s,%s)", ("book", "数据库📘", 5))
    query("INSERT INTO products(sku,name,stock) VALUES(%s,%s,%s)", ("pen", "笔", 1))
    assert query("SELECT name,note FROM products WHERE sku='book'") == [("数据库📘", None)]
    assert query("SELECT COUNT(*) FROM products WHERE note=NULL")[0][0] == 0
    assert query("SELECT COUNT(*) FROM products WHERE note IS NULL")[0][0] == 2
    assert attempt("first", "book", 2)["replayed"] is False
    assert attempt("first", "book", 2)["replayed"] is True
    errors = {"payload": expect(ValueError, lambda: attempt("first", "book", 1)),
              "stock": expect(ValueError, lambda: attempt("too-many", "book", 9)),
              "rollback": expect(RuntimeError, lambda: attempt("fail", "book", 1, True)),
              "injection": expect(ValueError, lambda: attempt("injection", "x' OR 1=1 --", 1)),
              "check": expect(mysql.connector.Error, lambda: query("UPDATE products SET stock=-1 WHERE sku='book'"))}
    assert query("SELECT stock FROM products WHERE sku='book'") == [(3,)]
    assert query("SELECT COUNT(*) FROM reservations") == [(1,)]
    owner = connect()
    try:
        c = owner.cursor()
        c.execute("UPDATE products SET stock=stock+1 WHERE sku='book'")
        expect(ValueError, lambda: reserve(owner, "owned", "book", 1))
        c.execute("SELECT stock FROM products WHERE sku='book'")
        assert c.fetchone() == (4,), "must preserve caller transaction"
    finally:
        owner.rollback()
        owner.close()
    barrier = threading.Barrier(2)
    def contender(key):
        barrier.wait(timeout=10)
        try:
            attempt(key, "pen", 1)
            return "reserved"
        except ValueError as error:
            return str(error)
    with ThreadPoolExecutor(max_workers=2) as pool:
        outcomes = list(pool.map(contender, ["race-a", "race-b"]))
    assert sorted(outcomes) == ["insufficient stock", "reserved"]
    assert query("SELECT stock FROM products WHERE sku='pen'") == [(0,)]
    assert query("SELECT p.sku,COALESCE(SUM(r.quantity),0) FROM products p LEFT JOIN reservations r ON r.sku=p.sku GROUP BY p.sku ORDER BY p.sku") == [("book", 2), ("pen", 1)]
    assert query("SELECT sku FROM reservations GROUP BY sku HAVING SUM(quantity)>=2") == [("book",)]
    # A real lock timeout: another independent connection holds the row lock.
    holder = connect()
    held = holder.cursor()
    held.execute("SELECT stock FROM products WHERE sku='book' FOR UPDATE")
    held.fetchall()
    waiter = connect()
    try:
        c = waiter.cursor()
        c.execute("SET SESSION innodb_lock_wait_timeout=1")
        error = expect(mysql.connector.Error, lambda: reserve(waiter, "timeout", "book", 1))
        assert "1205" in error
        errors["lock_timeout"] = error
    finally:
        waiter.close()
        holder.rollback()
        holder.close()
    cli = subprocess.run([sys.executable, str(Path(__file__).with_name("inventory.py")),
                          "reserve", "bad-cli", "book", "0"], capture_output=True, text=True)
    assert cli.returncode == 1 and "quantity" in cli.stderr
    success = subprocess.run([sys.executable, str(Path(__file__).with_name("inventory.py")),
                              "reserve", "first", "book", "2"], capture_output=True, text=True)
    assert success.returncode == 0 and json.loads(success.stdout)["replayed"] is True
    plan = query("EXPLAIN SELECT * FROM reservations WHERE sku='book' ORDER BY created_at")
    # EXPLAIN column 6 is the selected key; possible_keys alone is not evidence.
    selected_keys = [row[6] for row in plan]
    analyzed = query("EXPLAIN ANALYZE SELECT * FROM reservations WHERE sku='book' ORDER BY created_at")
    assert "actual time" in str(analyzed)
    output = {"server": query("SELECT VERSION(),@@version_compile_os,@@transaction_isolation,@@character_set_connection"),
              "connector": mysql.connector.__version__, "python": sys.version,
              "checks": ["utf8mb4", "NULL", "parameter binding", "idempotency", "payload conflict", "CHECK constraint", "rollback", "caller transaction preserved", "two-connection race", "JOIN/HAVING", "lock timeout", "CLI exit 1", "EXPLAIN", "EXPLAIN ANALYZE"],
              "errors": errors, "race": outcomes, "explain": plan, "selectedKeys": selected_keys, "analyzed": analyzed,
              "cli": {"exit": cli.returncode, "stderr": cli.stderr.strip(),
                      "successExit": success.returncode, "success": json.loads(success.stdout)},
              "products": query("SELECT sku,stock FROM products ORDER BY sku")}
    print(json.dumps(output, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

运行verify预期返回JSON并退出0：checks包含utf8mb4、NULL、参数绑定、幂等、CHECK、回滚、独立连接竞争、锁超时、CLI exit1和EXPLAIN；race数组一项reserved一项insufficient stock，顺序不固定。selectedKeys记录EXPLAIN实际选择的key，可能为null；小表全扫不是功能失败，不能把possible_keys出现索引当作已选用。3819是数据库CHECK违反，1205是真实锁超时。任何断言失败、连不上服务或缺驱动都会非零退出，不跳过后声称通过。

Linux实测使用官方python:3.12.14-slim一次性容器挂载此目录，在容器内安装requirements并运行相同单测与verify，通过host.docker.internal:23306连接Linux MySQL服务；Windows实测使用CPython3.12.14虚拟环境。环境与输出见迁移证据，未把跨平台代码本身当运行证明。

备份恢复完整脚本 ​

这份Python脚本调用官方容器里的mysqldump/mysql，用subprocess字节输入输出避开Windows PowerShell5重定向编码问题。它固定访问showdocs-m4-mysql-server容器，不读取CLI的MYSQL_HOST等连接变量。它只创建inventory_restore，若已存在则失败，不自动覆盖；但每次都会先覆盖本目录inventory-backup.sql，需要保留旧文件时先另存。详细使用与恢复边界见备份恢复。实验root密码在脚本中仅对应教程容器，生产脚本须通过受控配置提供凭据。

py
"""Logical backup and restore into a separate empty lab database via Docker."""
import hashlib
import json
from pathlib import Path
import subprocess

CONTAINER = "showdocs-m4-mysql-server"


def docker(*args, data=None):
    command = ["docker", "exec", "-i", "-e", "MYSQL_PWD=m4-local-root", CONTAINER, *args]
    result = subprocess.run(command, input=data, capture_output=True, check=True)
    return result.stdout


def main():
    # No shell redirection: preserve dump bytes on Windows and Linux alike.
    dump = docker("mysqldump", "-uroot", "--default-character-set=utf8mb4", "--single-transaction", "--no-tablespaces",
                  "--set-gtid-purged=OFF", "--skip-comments", "inventory_lab")
    backup = Path(__file__).with_name("inventory-backup.sql")
    backup.write_bytes(dump)
    # Refuse to overwrite an existing restore database. Explicitly remove only
    # inventory_restore yourself before repeating this exercise.
    docker("mysql", "-uroot", "-e", "CREATE DATABASE inventory_restore CHARACTER SET utf8mb4")
    docker("mysql", "-uroot", "--default-character-set=utf8mb4", "inventory_restore", data=dump)
    # Match the dump's UTC timestamp convention and include every stored column.
    sql = "SET time_zone='+00:00'; SELECT sku,name,stock,note FROM products ORDER BY sku; SELECT request_id,sku,quantity,created_at FROM reservations ORDER BY request_id"
    source = docker("mysql", "-uroot", "--default-character-set=utf8mb4", "-N", "inventory_lab", "-e", sql)
    restored = docker("mysql", "-uroot", "--default-character-set=utf8mb4", "-N", "inventory_restore", "-e", sql)
    assert source == restored and "数据库📘" in restored.decode()
    print(json.dumps({"dumpBytes": len(dump), "sha256": hashlib.sha256(dump).hexdigest(),
                      "restored": restored.decode(), "equal": True}, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

研究依据：Connector/Python参数执行、事务启动。

自检练习：题4至6。提示：分别观察request_id、产品锁及调用前in_transaction；解答另列。

继续阅读