MySQL 完整应用：Python 库存预留 CLI · 项目结构与运行至完整 CLI 与事务合同

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

MySQL 完整应用：Python 库存预留 CLI

项目结构与运行

首次运行只需完成环境准备，可先初始化，再返回建模与事务解释；阅读实现前应理解事务与并发。项目只提供list和reserve命令，没有HTTP服务。文件位于仓库examples/mysql/inventory；本页直接包含相同源码，修改示例后正文同步展示，避免两份实现漂移。依赖、SQL、CLI、单测、数据库验收、恢复六份源文件构成完整项目；另有历史生成的inventory-backup.sql，它不是初始化输入，不能直接导入已有库。Python3.12+，官方驱动固定9.5.0，MySQL8.4.7，服务准备见起步。

txt

mysql-connector-python==9.5.0

PowerShell在仓库根目录运行；venv隔离本站依赖，不需要激活：

powershell

python -m venv examples/mysql/inventory/.venv
examples/mysql/inventory/.venv/Scripts/python.exe -m pip install -r examples/mysql/inventory/requirements.txt
$env:MYSQL_PASSWORD='m4-local-lab'
examples/mysql/inventory/.venv/Scripts/python.exe -m unittest discover -s examples/mysql/inventory -p test_inventory.py
examples/mysql/inventory/.venv/Scripts/python.exe -X utf8 examples/mysql/inventory/verify.py
examples/mysql/inventory/.venv/Scripts/python.exe -X utf8 examples/mysql/inventory/inventory.py list
examples/mysql/inventory/.venv/Scripts/python.exe -X utf8 examples/mysql/inventory/inventory.py reserve user-1 book 1

Bash路径不同，其余SQL/业务行为共用：

python3 -m venv examples/mysql/inventory/.venv
examples/mysql/inventory/.venv/bin/python -m pip install -r examples/mysql/inventory/requirements.txt
export MYSQL_PASSWORD=m4-local-lab
examples/mysql/inventory/.venv/bin/python -m unittest discover -s examples/mysql/inventory -p test_inventory.py
examples/mysql/inventory/.venv/bin/python examples/mysql/inventory/verify.py
examples/mysql/inventory/.venv/bin/python examples/mysql/inventory/inventory.py list
examples/mysql/inventory/.venv/bin/python examples/mysql/inventory/inventory.py reserve user-1 book 1

verify会创建两张表，删除inventory_lab中的这两张表的实验数据再初始化，只能对专用实验库运行。verify在修改数据前要求MYSQL_DATABASE为inventory_lab；connect仅允许inventory_lab/inventory_restore库名，但名称检查不能识别同名业务库，仍须核对主机、端口与账号。若解释器裁剪了ensurepip，改用带pip的Python创建虚拟环境后重试完整安装命令。此前Windows记录采用另一个有pip的解释器向该虚拟环境安装，未改系统Python依赖。

verify后list返回book库存3、pen库存0及中文名；reserve user-1成功返回replayed=false，再执行同参数返回true。相同ID改数量、库存不足、未知sku、零或负数量均输出真实错误并退出1；argparse语法错误退出2。演示后重跑verify会重置实验数据。MYSQL_HOST默认127.0.0.1，MYSQL_PORT默认23306，MYSQL_USER默认lab；密码无默认值，缺少即报KeyError并非零退出。

完整 CLI 与事务合同

初始化器从自身目录读取以下完整DDL，表结构在建模解释同源展示。CREATE TABLE IF NOT EXISTS不会升级旧表；已有表与DDL不一致时应另建干净实验环境，不要把verify当迁移工具。

sql

CREATE TABLE IF NOT EXISTS products (
  sku VARCHAR(64) COLLATE utf8mb4_bin PRIMARY KEY,
  name VARCHAR(120) NOT NULL,
  stock INT NOT NULL CHECK (stock >= 0),
  note VARCHAR(120) NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS reservations (
  request_id VARCHAR(64) COLLATE utf8mb4_bin PRIMARY KEY,
  sku VARCHAR(64) COLLATE utf8mb4_bin NOT NULL,
  quantity INT NOT NULL CHECK (quantity > 0),
  created_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  FOREIGN KEY (sku) REFERENCES products(sku),
  INDEX ix_sku_created (sku, created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

reserve要求无活跃事务的独立连接；若调用方已有未提交事务，函数在开始前拒绝，并保留调用方修改。函数仅负责自己开启的事务，不支持嵌套，也不调用外部HTTP。正常返回前commit，异常路径rollback，finally关闭游标，CLI最终关闭连接。MySQL 参数绑定由驱动游标以%s占位并单独传元组完成，不能用f-string或拼接用户输入构建SQL；标识符表名不能通过值占位符绑定，本项目固定表名。

"""An isolated inventory reservation CLI; no HTTP server or implicit retries."""
import argparse
import json
import os
import sys


def validate(request_id, sku, quantity):
    if any(type(value) is not str or not 1 <= len(value) <= 64 or value.endswith(" ")
           for value in (request_id, sku)):
        raise ValueError("request_id and sku must contain 1..64 characters without trailing spaces")
    if type(quantity) is not int or not 1 <= quantity <= 2147483647:
        raise ValueError("quantity must be an integer in 1..2147483647")


def connect():
    import mysql.connector
    database = os.environ.get("MYSQL_DATABASE", "inventory_lab")
    if database not in ("inventory_lab", "inventory_restore"):
        raise ValueError("only isolated inventory_lab/inventory_restore allowed")
    return mysql.connector.connect(
        host=os.environ.get("MYSQL_HOST", "127.0.0.1"),
        port=int(os.environ.get("MYSQL_PORT", "23306")),
        user=os.environ.get("MYSQL_USER", "lab"),
        password=os.environ["MYSQL_PASSWORD"], database=database,
        charset="utf8mb4", autocommit=False, connection_timeout=5,
        use_pure=True)


def reserve(db, request_id, sku, quantity, fail_after_update=False):
    validate(request_id, sku, quantity)
    if db.in_transaction:
        raise ValueError("reserve requires a connection without an active transaction")
    cursor = db.cursor()
    try:
        db.start_transaction(isolation_level="READ COMMITTED")
        # Every reservation locks the product first, including idempotent retries.
        cursor.execute("SELECT stock FROM products WHERE sku=%s FOR UPDATE", (sku,))
        product = cursor.fetchone()
        if product is None:
            raise ValueError("unknown sku")
        cursor.execute("SELECT sku, quantity FROM reservations WHERE request_id=%s",
                       (request_id,))
        previous = cursor.fetchone()
        if previous:
            if previous != (sku, quantity):
                raise ValueError("idempotency key reused with different payload")
            db.commit()
            return {"request_id": request_id, "replayed": True}
        if product[0] < quantity:
            raise ValueError("insufficient stock")
        cursor.execute("UPDATE products SET stock=stock-%s WHERE sku=%s", (quantity, sku))
        if fail_after_update:
            raise RuntimeError("injected failure after decrement")
        cursor.execute("INSERT INTO reservations(request_id,sku,quantity) VALUES(%s,%s,%s)",
                       (request_id, sku, quantity))
        db.commit()
        return {"request_id": request_id, "replayed": False}
    except Exception:
        db.rollback()
        raise
    finally:
        cursor.close()


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="command", required=True)
    sub.add_parser("list")
    reservation = sub.add_parser("reserve")
    reservation.add_argument("request_id")
    reservation.add_argument("sku")
    reservation.add_argument("quantity", type=int)
    args = parser.parse_args()
    db = None
    try:
        db = connect()
        if args.command == "reserve":
            result = reserve(db, args.request_id, args.sku, args.quantity)
        else:
            cursor = db.cursor()
            cursor.execute("SELECT sku,name,stock,note FROM products ORDER BY sku")
            result = cursor.fetchall()
            cursor.close()
        print(json.dumps(result, ensure_ascii=False))
        return 0
    except Exception as error:
        print(f"{type(error).__name__}: {error}", file=sys.stderr)
        return 1
    finally:
        if db is not None:
            db.close()


if __name__ == "__main__":
    sys.exit(main())

输入合同要求request_id与sku都是1至64字符的字符串，并拒绝末尾ASCII空格。DDL使用的utf8mb4_bin是PAD SPACE排序规则，二进制比较仍可能忽略末尾空格；拒绝这类输入可避免Python字符串比较与数据库唯一键判断不一致。单测中已加入类型、长度、数量上界与尾空格反例，本轮未运行。直接SQL写入方也须遵守同一标识合同。

请求键永久保留在实验表，幂等保证限于记录仍存在且事务落在同一个库。若以后归档/删除，必须先定义重试有效期及归档查询策略。这里无连接池、自动重试、预留取消或补货命令；这些不是库存预留原子性成立的前提。CLI输出的是执行结果，不能把日志当业务账本。

MySQL 完整应用：Python 库存预留 CLI · 项目结构与运行至完整 CLI 与事务合同 ​