ShowDocs
搜索文章Ctrl K

外观

← 返回MongoDB列表
应用实践

MongoDB：版本化任务完整 CLI ​

前置是文档过滤器、单文档原子性和环境变量。适用 MongoDB Server 8.0 版本族及 PyMongo >=4.10,<5，核对 2026-10-17。示例未执行，依赖未安装；未连接任何数据库、云账户或模拟器。

配置、依赖与命令顺序 ​

MongoDB 任务 CLI 使用 MONGODB_URI 与 MONGODB_DATABASE 配置连接，.env.example 只是阅读样本，Python 不会自动加载它。完整文件只有 tasks.py、requirements.txt 和 .env.example。未来读者需自行准备被授权的现有副本集、具备集合读写和建索引权限的用户以及兼容 Python 3.12+ 环境；这里不启动或部署服务器。

PowerShell 设置环境变量采用 $env:MONGODB_URI='你的已批准连接串'；Linux shell 使用 export MONGODB_URI='你的已批准连接串'。不要把实际密钥提交进源码或命令记录。数据库默认 theory_tasks，可单独设置 MONGODB_DATABASE。依赖声明见下方；读者自行安装时使用其虚拟环境的 python -m pip install -r requirements.txt，本轮未执行。

text
python tasks.py init
python tasks.py add t1 alice
python tasks.py finish t1 0
python tasks.py finish t1 0
python tasks.py list
python tasks.py report

预期 init 输出 index=owner_status_id；add 输出 id=t1,version=0；第一次 finish 返回 status=done,version=1；第二次 finish 未匹配并退出 3。list 返回最多 50 项且按 _id 排序；report 返回 done 的 count=1。重复 add 的 _id 冲突应退出 1，不会覆盖任务。本示例没有删除命令，后续清理应由读者确认独立教学数据库后另行进行。

条件更新的完整源码 ​

MongoDB 条件更新把版本和状态放进过滤器，$inc 与 $set 在同一单文档操作中生效。find_one_and_update 用 ReturnDocument.AFTER 返回更新后文档；None 表示未匹配，不能当作完成成功。用户输入只能填入固定字段值，程序不接受任意 JSON 过滤器，避免把输入误解释为查询操作符。

py
"""Single-document compare-and-set CLI; not executed in this delivery."""
import argparse
import json
import os
import sys

from pymongo import ASCENDING, MongoClient, ReturnDocument
from pymongo.errors import PyMongoError
from pymongo.read_concern import ReadConcern
from pymongo.write_concern import WriteConcern


def main():
    parser = argparse.ArgumentParser(description="Versioned task transitions")
    sub = parser.add_subparsers(dest="command", required=True)
    sub.add_parser("init")
    add = sub.add_parser("add")
    add.add_argument("id")
    add.add_argument("owner")
    finish = sub.add_parser("finish")
    finish.add_argument("id")
    finish.add_argument("version", type=int)
    sub.add_parser("list")
    sub.add_parser("report")
    args = parser.parse_args()
    client = None
    try:
        uri = os.environ.get("MONGODB_URI")
        database = os.environ.get("MONGODB_DATABASE", "theory_tasks")
        if not uri or database in ("admin", "local", "config"):
            raise ValueError("set MONGODB_URI and a non-system MONGODB_DATABASE")
        if args.command == "add" and (not args.id or not args.owner):
            raise ValueError("id and owner required")
        if args.command == "finish" and (not args.id or args.version < 0):
            raise ValueError("id required; version must be nonnegative")
        client = MongoClient(uri, serverSelectionTimeoutMS=3000, connectTimeoutMS=3000, socketTimeoutMS=5000)
        collection = client[database].get_collection("tasks", read_concern=ReadConcern("majority"), write_concern=WriteConcern(w="majority", wtimeout=5000))
        if args.command == "init":
            name = collection.create_index([("owner", ASCENDING), ("status", ASCENDING), ("_id", ASCENDING)], name="owner_status_id")
            result = {"index": name}
        elif args.command == "add":
            collection.insert_one({"_id": args.id, "owner": args.owner, "status": "open", "version": 0})
            result = {"id": args.id, "version": 0}
        elif args.command == "finish":
            result = collection.find_one_and_update({"_id": args.id, "version": args.version, "status": "open"}, {"$set": {"status": "done"}, "$inc": {"version": 1}}, return_document=ReturnDocument.AFTER)
            if result is None:
                print("missing task, stale version or already done", file=sys.stderr)
                return 3
        elif args.command == "list":
            result = list(collection.find({}, {"owner": 1, "status": 1, "version": 1}).sort("_id", ASCENDING).limit(50))
        else:
            result = list(collection.aggregate([{"$group": {"_id": "$status", "count": {"$sum": 1}}}, {"$sort": {"_id": 1}}]))
        print(json.dumps(result, ensure_ascii=False))
        return 0
    except (PyMongoError, ValueError) as exc:
        # Driver exceptions can include hostnames; never print URI/credentials.
        print(type(exc).__name__ + ": operation failed; inspect configuration or database diagnostics", file=sys.stderr)
        return 1
    finally:
        if client is not None:
            client.close()


if __name__ == "__main__":
    sys.exit(main())
txt
pymongo>=4.10,<5
text
# Documentation only: Python does not automatically load this file.
# Export these variables in your shell; replace the local example with your approved target.
MONGODB_URI=mongodb://127.0.0.1:27017/?replicaSet=rs0
MONGODB_DATABASE=theory_tasks

驱动负责 BSON 编码，程序选择字符串 _id 以便使用标准 JSON 输出。超时配置限制连接选择和网络等待，不是总业务截止时间保证。连接在 finally 中关闭，进程不会以一个未关闭游标代替持久会话。

错误与退出码 ​

MongoDB 退出码约定为成功 0、配置或驱动异常 1、参数格式错误 2、条件未匹配 3。版本过期、任务缺失和已经完成统一返回 3，若业务要区分可另读状态，但那次读取与之前失败之间仍有竞态。重复 _id 会是 DuplicateKeyError；连接不可达可能是 ServerSelectionTimeoutError；认证权限错误需要核对 authSource 和角色。

程序不打印原 URI、密码或原始驱动错误，以免泄露连接信息；实际排错由有权限的操作者查看脱敏诊断。驱动异常中的写关注超时可能留下已写入但确认不完整的结果，不能自动重复 add 并称为绝对安全；应先按 _id 查询。CLI 未实现端到端请求幂等协议，也没有多文档事务。

阶段练习与参考方案 ​

MongoDB 实践要求两位操作者都以 version=0 完成 t1，预期最多一个匹配，另一个退出 3；本轮只做理论推演，未执行并发。再设计 owner=alice 且 status=open 的查询，明确使用 owner_status_id 的前缀和 _id 顺序。

参考方案：固定过滤器包含 owner 和 status，sort _id，limit 20；用 explain 检查计划，不能硬编码预期一定 IXSCAN。报告按状态计数，保留 $sort 使输出稳定。若想加入审计历史，需要重新设计增长边界和跨文档一致性，而不是无界追加一个数组。

继续阅读