ShowDocs
搜索文章Ctrl K

外观

← 返回PostgreSQL列表
应用实践

PostgreSQL 完整项目：参数化库存预留 CLI · 安装、初始化与运行至完整源码与参数绑定 ​

返回分篇目录

PostgreSQL 完整项目：参数化库存预留 CLI ​

本轮仅做正文和源码静态审阅，以下命令、SQL、集成测试及恢复流程均未执行。过去真实运行记录保留在 migration/knowledge-rebuild/m4-next/postgres/verification.md 及其原始日志；历史结果不证明修改后的示例运行通过。

安装、初始化与运行 ​

环境前置：按起步配置准备隔离数据库；首次学习到起步 SQL 时，可以只完成本节初始化、verify 和 list，再返回原处。理解完整实现的知识前置是事务与所有权。这个独立教学CLI有init、list、reserve三个子命令，没有给本站增加后端。所有文件位于仓库 examples/postgres/：compose.yaml、requirements.txt、schema.sql、inventory.py、test_inventory.py、verify.py、red_check.py、backup_restore.py。本文用源码include展示同一文件，避免复制版本漂移。Python3.12+，Psycopg3.2.13支持对应Python和PostgreSQL，固定教学版本而非宣称最新驱动。

txt
psycopg[binary]==3.2.13

PowerShell在仓库根目录运行，使用独立venv，不升级本站依赖：

powershell
python -m venv examples/postgres/.venv
examples/postgres/.venv/Scripts/python.exe -m pip install -r examples/postgres/requirements.txt
$env:PGPASSWORD='m4-local-lab'
examples/postgres/.venv/Scripts/python.exe -X utf8 examples/postgres/inventory.py init
examples/postgres/.venv/Scripts/python.exe -X utf8 examples/postgres/verify.py
examples/postgres/.venv/Scripts/python.exe -X utf8 examples/postgres/inventory.py list
examples/postgres/.venv/Scripts/python.exe -X utf8 examples/postgres/inventory.py reserve user-1 book 1

Bash使用对应虚拟环境路径，SQL和源码完全相同：

sh
python3 -m venv examples/postgres/.venv
examples/postgres/.venv/bin/python -m pip install -r examples/postgres/requirements.txt
export PGPASSWORD=m4-local-lab
examples/postgres/.venv/bin/python examples/postgres/inventory.py init
examples/postgres/.venv/bin/python examples/postgres/verify.py
examples/postgres/.venv/bin/python examples/postgres/inventory.py list
examples/postgres/.venv/bin/python examples/postgres/inventory.py reserve user-1 book 1

若解释器不带 ensurepip，历史交付曾使用带pip的解释器执行 python -m pip --python examples/postgres/.venv/Scripts/python.exe install -r examples/postgres/requirements.txt，依然安装到隔离环境。PGHOST默认127.0.0.1、PGPORT默认25432、用户名固定lab，PGPASSWORD必填；connect仅接受postgres_lab/postgres_restore库名。库名白名单不校验实例身份，遗留 PGHOST/PGPORT 仍可能把操作指向其他实例；执行前用起步 SQL 核对连接目标，并显式使用本机 127.0.0.1:25432。示例不能指向业务库。verify和测试都会 TRUNCATE这两张实验表，只对本机独立实例执行。

verify后 list 按sku返回book/中文📚/3、pen/笔/0。第一次 reserve user-1 book 1 返回 replayed=false 和stock=2；重复同参数返回replayed=true，不再扣减。再次执行同键但数量2，返回“idempotency key reused with different parameters”且退出1。pen库存0、未知sku、数量0/负数都失败且不留记录；未知sku触发外键错误。argparse用法错误退出2，业务/连接错误退出1，成功退出0。

init只创建表，不填种子数据。verify填入的seed预留是为了恢复时有意义地检查两张表；重复verify会重置演示状态。仅list成功不能证明reserve的回滚和并发正确，下面测试分别验证。

完整源码与参数绑定 ​

py
"""Small parameterized inventory CLI. Only connects to the dedicated lab databases."""
import argparse
import json
import os
from pathlib import Path
import sys

import psycopg


def connect(database="postgres_lab"):
    if database not in {"postgres_lab", "postgres_restore"}:
        raise ValueError("only dedicated lab databases are allowed")
    return psycopg.connect(host=os.getenv("PGHOST", "127.0.0.1"),
                           port=int(os.getenv("PGPORT", "25432")),
                           user="lab", password=os.environ["PGPASSWORD"],
                           dbname=database, autocommit=True, connect_timeout=5)


def initialize(conn):
    conn.execute(Path(__file__).with_name("schema.sql").read_text(encoding="utf-8"))


def reserve(conn, request_id, sku, quantity):
    if not request_id or not sku or type(quantity) is not int or quantity <= 0:
        raise ValueError("request_id/sku required; quantity must be a positive integer")
    # Own transaction when idle; a SAVEPOINT when the caller owns a transaction.
    with conn.transaction():
        inserted = conn.execute(
            "INSERT INTO reservations(request_id, sku, quantity) VALUES (%s,%s,%s) "
            "ON CONFLICT (request_id) DO NOTHING RETURNING request_id",
            (request_id, sku, quantity)).fetchone()
        if inserted is None:
            previous = conn.execute(
                "SELECT sku, quantity FROM reservations WHERE request_id=%s",
                (request_id,)).fetchone()
            if previous != (sku, quantity):
                raise ValueError("idempotency key reused with different parameters")
            return {"request_id": request_id, "replayed": True}
        updated = conn.execute(
            "UPDATE products SET stock=stock-%s WHERE sku=%s AND stock >= %s "
            "RETURNING stock", (quantity, sku, quantity)).fetchone()
        if updated is None:
            raise ValueError("unknown sku or insufficient stock")
        return {"request_id": request_id, "replayed": False, "stock": updated[0]}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    commands = parser.add_subparsers(dest="command", required=True)
    commands.add_parser("init")
    commands.add_parser("list")
    reserve_parser = commands.add_parser("reserve")
    reserve_parser.add_argument("request_id")
    reserve_parser.add_argument("sku")
    reserve_parser.add_argument("quantity", type=int)
    args = parser.parse_args()
    try:
        with connect() as conn:
            if args.command == "init":
                initialize(conn)
                result = {"initialized": True}
            elif args.command == "list":
                result = conn.execute("SELECT sku,name,stock FROM products ORDER BY sku").fetchall()
            else:
                result = reserve(conn, args.request_id, args.sku, args.quantity)
        print(json.dumps(result, ensure_ascii=False))
        return 0
    except (ValueError, psycopg.Error, KeyError) as error:
        print(str(error), file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

SQL值一律通过 %s 占位符及独立参数元组传给 execute，不能用Python的 %、f-string或拼接把用户输入塞入SQL字符串。占位符外不加引号，字符串、整数和特殊字符由驱动编码。占位符不能替代表名和排序方向，这类动态标识符需要白名单或psycopg.sql.Identifier；本CLI使用固定SQL，不引入无用动态表名能力。

连接上下文保证关闭连接；reserve自己的事务或SAVEPOINT包住两条写语句。返回JSON是客户端确认信息，真正的幂等依据在服务端主键记录。进程在提交后、打印前崩溃时客户端可能不知道结果，使用同一request_id重试能查询已提交结果；每次重试生成新键会失去这个性质。删除幂等记录会改变重试语义，本例不提供清理策略。

继续阅读