AI 技术博客
返回首页
Python 基础 28 分钟阅读

实战:待办事项命令行工具

Python命令行工具argparseJSON实战

引言

这是第九章「综合项目:命令行工具」的第 8 篇,也是本章的收官工程篇。前七篇的零件已经全部到位:第 1 篇的工程骨架、第 2、3 篇的 argparse(位置参数、类型转换、子命令、自动 help)、第 4、5 篇的 logging、第 6 篇的退出码与机器可读输出、第 7 篇的单元测试。本篇要把这些零件拧成一台完整的机器——待办事项命令行工具(todo CLI):用 todo add 添加任务、todo list 查看、todo done 标记完成、todo remove 删除,任务数据持久化在 JSON 文件里,关掉终端、重启电脑都还在。

为什么选待办工具作为收官实战?因为它是增查改删 + 持久化(CRUD + persistence)这一真实工程最小闭环的样板:增删改查四个操作、数据要落盘、错误要让用户看懂、逻辑要经得起测试。任何更复杂的工具(记账、笔记、任务调度)都是这个骨架的加厚。全部实现只依赖标准库:argparse + json + logging + pathlib,不装任何第三方包——这本身就是”命令行工具”的正确起点:小工具零依赖,用得舒服了再谈打包分发(那是第 9 篇的事)。

读者应先掌握:第 3 篇的 add_subparsers 子命令、第七章的 json 读写与异常捕获(json.dump/load、JSONDecodeError)、第 4-5 篇的 logging、第 6 篇的 sys.exit 退出码、上一篇的 unittest 测试方法。本篇的工程结构将直接成为第 9 篇打包的对象,而第 10 篇测验会考察本篇的每个决策点,请务必亲手运行每一个代码块。

概念与原理

分层:存储层与命令层

小脚本最常见的腐化方式,是”一个 main 函数里什么都干”:解析参数、读文件、改数据、写文件、打印全搅在一起。数据格式一改,命令逻辑跟着动;命令加一个,文件读写重写一遍。本篇采用两层的职责分离:

  • 存储层(storage layer):只负责”任务数据怎么进、怎么出”——从 JSON 文件读回任务列表、把任务列表写进 JSON 文件、以及增删改查四个操作。它完全不关心用户是敲命令还是调 API;
  • 命令层(command layer):只负责”用户怎么用”——argparse 解析命令行、调用存储层、按结果打印与决定退出码。它不直接碰文件。

分层的直接红利在测试上:上一篇讲过”测试只能测没有 I/O 纠缠的单元”,而存储层把文件 I/O 收敛进自己的方法后,命令层即便不被自动化测试覆盖,存储层也能被 unittest 完整测透。另外,后续给工具加 GUI、加 Web 接口时,命令层可以整个换掉,存储层原封不动——这就是单一职责(single responsibility)在架构上的回报。

为什么 JSON:有类型、可读、标准库

任务的形状是”编号、文本、完成标记”三条字段的列表,这正是第七章讲过的结构化数据。选 JSON 的理由:其一,类型自带——done 字段存成真布尔 true/false,读回就是 Python 的 True/False,不用像 CSV 那样全部转字符串(第七章第 7 篇的对比结论);其二,人可读且可手工编辑——用文本编辑器打开 todo.json 就能改,这对一个个人工具是隐性需求;其三,标准库直读——json.dump/load 的用法第七章已经全部讲过,零学习成本。量级上,个人的待办列表撑死几千条,json.load 一次全量读入内存完全无压力;若未来数据量到百万级,再考虑 sqlite(数据库,课程 05 的内容),现在不做过度设计。

原子写:写文件也要防”半截”

直接 open(path, 'w') 写 JSON 有一个隐患:写入过程被中断(断电、崩溃、误触 Ctrl-C),文件里可能只剩半个 JSON 结构,下次启动 json.load 直接炸。原子写(atomic write)是标准解法:先把内容写进同目录的临时文件,再用 Path.replace 一次性替换目标文件。replace 在 POSIX 上对应 rename(2),是文件系统的原子操作——要么旧的完整文件在,要么新的完整文件在,绝不存在”中间态”。两个要点:临时文件必须与目标同目录(跨目录/跨磁盘的 rename 可能失败或非原子);命名要带 .tmp 后缀避免与真实文件混淆。本章的存储层把原子写沉淀进 save 方法,损坏兜底(易错点第 3 条)就是为它准备的第二道保险。

编号策略:只增不减

任务编号用 max(现有编号) + 1 分配、删除后不回收编号。为什么不为”编号连续”而把后面的任务全部重排?因为编号对外是身份标识:某人已经记住了”#3 是写报告”,删除 #2 后把 #3 改成 #2,会造成引用错乱和同步工具的困惑。真实系统里自增主键(如 SQLite 的 AUTOINCREMENT)也是这个语义——编号是身份,不是序号。这也是 done 9(不存在的编号)必须明确报错的根源:id 语义下,非法引用要出声,不能静默。

退出码与日志:给机器和人都留下痕迹

参照第 6 篇的约定:0 表示成功;业务错误(任务不存在)返回 1;参数用错(如 todo done 不带编号)由 argparse 自身以退出码 2 报错——这是 argparse 的既定行为,三个退出码把”成功/业务失败/用法错误”区分开,脚本化调用时一目了然。日志方面,命令层用 logging 的 error 级别输出错误(第 4、5 篇的配置方式),与正常 stdout 输出分流:用户看 stdout 的”已完成 #2”,出问题时 stderr 上有 ERROR 行,两边都不互相污染。

操作与实现

第 1 步:存储层 todo_store.py

先写存储层。注意 load 方法的三条路径:文件不存在返回空列表(首次运行不报错)、正常解析返回列表、JSON 损坏则记录错误、把损坏文件改名备份后重新开始。save 方法实现原子写。保存为 todo_store.py:

# todo_store.py:待办工具的存储层(JSON 持久化 + 原子写)
import json
import logging
from pathlib import Path

LOG = logging.getLogger('todo.store')

class TaskStore:
    """把任务列表持久化到单个 JSON 文件。"""

    def __init__(self, path):
        self.path = Path(path)

    def load(self):
        """读回全部任务;文件不存在返回空列表,损坏时备份后重新开始。"""
        if not self.path.exists():
            return []
        try:
            return json.loads(self.path.read_text(encoding='utf-8'))
        except json.JSONDecodeError:
            LOG.error('存储文件损坏:%s 已改名备份,重新开始', self.path)
            backup = self.path.with_name(self.path.stem + '.corrupted.json')
            self.path.replace(backup)
            return []

    def save(self, tasks):
        """原子写:先写临时文件再整体替换,避免写一半留下坏文件。"""
        tmp = self.path.with_name(self.path.stem + '.tmp.json')
        tmp.write_text(json.dumps(tasks, ensure_ascii=False, indent=2),
                       encoding='utf-8')
        tmp.replace(self.path)

    def add(self, text):
        """新增任务,编号为现有最大编号 + 1(编号不重用)。"""
        tasks = self.load()
        new_id = max((t['id'] for t in tasks), default=0) + 1
        tasks.append({'id': new_id, 'text': text, 'done': False})
        self.save(tasks)
        return new_id

    def list_all(self):
        return self.load()

    def mark_done(self, task_id):
        """标记完成;编号不存在返回 False。"""
        tasks = self.load()
        for t in tasks:
            if t['id'] == task_id:
                t['done'] = True
                self.save(tasks)
                return True
        return False

    def remove(self, task_id):
        """删除任务;编号不存在返回 False。"""
        tasks = self.load()
        rest = [t for t in tasks if t['id'] != task_id]
        if len(rest) < len(tasks):
            self.save(rest)
            return True
        return False

if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
    store = TaskStore('demo_todo.json')
    store.save([])          # 每次演示从空开始,脚本可反复运行
    t1 = store.add('买菜')
    t2 = store.add('写单元测试')
    store.add('复习第七章')
    print('初始三件套:', len(store.list_all()), '件')
    store.mark_done(t2)
    store.remove(t1)
    print('完成一件、删掉一件后,还剩:')
    for t in store.list_all():
        print(' ', t)
    print('磁盘上的 demo_todo.json 内容:')
    print(store.path.read_text(encoding='utf-8'))

直接运行 python todo_store.py:

初始三件套: 3 件
完成一件、删掉一件后,还剩:
  {'id': 2, 'text': '写单元测试', 'done': True}
  {'id': 3, 'text': '复习第七章', 'done': False}
磁盘上的 demo_todo.json 内容:
[
  {
    "id": 2,
    "text": "写单元测试",
    "done": true
  },
  {
    "id": 3,
    "text": "复习第七章",
    "done": false
  }
]

几个细节对照原理部分:删除 #1 后 #2、#3 保持原编号(只增不减);ensure_ascii=False + encoding='utf-8' 让磁盘上是可读中文(第七章第 7 篇组合);落盘格式 indent=2 让人工编辑友好。__main__ 分支是给存储层的”自检演示”,真实使用的入口是命令层。

第 2 步:命令层 todo.py

命令层用第 3 篇的 add_subparsers 声明四个子命令,然后按 args.command 分发。注意 mark_done/remove 返回 False(编号不存在)时写 ERROR 日志并返回退出码 1——第 6 篇的约定在这里落地。保存为 todo.py(与 todo_store.py 同一目录):

# todo.py:命令行层,负责解析参数、调用存储层、决定退出码
import argparse
import logging
import sys
from todo_store import TaskStore

LOG = logging.getLogger('todo')

def build_parser():
    parser = argparse.ArgumentParser(prog='todo', description='待办事项命令行工具')
    sub = parser.add_subparsers(dest='command', required=True)
    for name, help_ in [('add', '新增待办'), ('list', '列出待办'),
                        ('done', '标记完成'), ('remove', '删除任务')]:
        sub.add_parser(name, help=help_)
    sub.choices['add'].add_argument('text', help='任务描述,含空格时请加引号')
    sub.choices['done'].add_argument('id', type=int, help='任务编号')
    sub.choices['remove'].add_argument('id', type=int, help='任务编号')
    return parser

def main(argv=None):
    logging.basicConfig(level=logging.WARNING, format='%(levelname)s: %(message)s')
    args = build_parser().parse_args(argv)
    store = TaskStore('todo.json')
    if args.command == 'add':
        print('已添加 #%d: %s' % (store.add(args.text), args.text))
    elif args.command == 'list':
        tasks = store.list_all()
        if not tasks:
            print('(空)还没有待办事项')
        for t in tasks:
            print('[%s] %d %s' % ('x' if t['done'] else ' ', t['id'], t['text']))
    elif args.command == 'done':
        if store.mark_done(args.id):
            print('已完成 #%d' % args.id)
        else:
            LOG.error('任务 #%d 不存在', args.id)
            return 1
    else:   # remove
        if store.remove(args.id):
            print('已删除 #%d' % args.id)
        else:
            LOG.error('任务 #%d 不存在', args.id)
            return 1
    return 0

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

命令层只有 47 行,因为它不碰任何数据细节——所有持久化都在 TaskStore 里。sub.choices 是 argparse 内部保存子解析器的字典,用它拿到刚注册的 parser 再补参数,比逐个变量保存省四行代码;第 3 篇还讲过,每个子命令现在自动获得 -h 帮助。跑一下 python todo.py --help,argparse 自动生成的用法一目了然:

usage: todo [-h] {add,list,done,remove} ...

待办事项命令行工具

positional arguments:
  {add,list,done,remove}
    add                 新增待办
    list                列出待办
    done                标记完成
    remove              删除任务

options:
  -h, --help            show this help message and exit

第 3 步:真实会话——把工具用起来

在命令行里完整走一遍用户视角的流程。假设上一段的两个文件已保存在当前目录:

# 清理旧数据,保证会话可重复运行(首次运行本就没有这些文件)
rm -f todo.json todo.corrupted.json
python todo.py add 买菜
python todo.py add "写单元测试"
python todo.py add 复习第七章
python todo.py list
python todo.py done 2
python todo.py list
python todo.py remove 3
python todo.py list
python todo.py done 9
echo "退出码: $?"
已添加 #1: 买菜
已添加 #2: 写单元测试
已添加 #3: 复习第七章
[ ] 1 买菜
[ ] 2 写单元测试
[ ] 3 复习第七章
已完成 #2
[ ] 1 买菜
[x] 2 写单元测试
[ ] 3 复习第七章
已删除 #3
[ ] 1 买菜
[x] 2 写单元测试
ERROR: 任务 #9 不存在
退出码: 1

注意三点:含空格的任务描述必须加引号(“写单元测试”),否则 argparse 只收到第一个词;done/remove 后列表保持原编号顺序,[x] 标记完成态;对不存在的 #9 执行 done,程序打印 ERROR: 任务 #9 不存在(走 logging 到 stderr)并以退出码 1 结束——echo $? 证实了这一点,这正是第 6 篇的”机器可读”语义。

再看看磁盘上的持久化结果。此刻 todo.json 的内容是:

[
  {
    "id": 1,
    "text": "买菜",
    "done": false
  },
  {
    "id": 2,
    "text": "写单元测试",
    "done": true
  }
]

id 3 被删除后没有重排,#2 依然指向”写单元测试”——编号身份语义落实到了字节级。这份文件可直接用文本编辑器修改,重启程序后 todo list 读到的就是你手改的内容。

第 4 步:摧毁与重建——损坏 JSON 的兜底演示

手工编辑可能改坏文件,写一半断电也会留下坏文件。用 printf 制造一份损坏的 JSON 再运行:

printf '{bad json' > todo.json   # 模拟文件被写坏
python todo.py list
echo "退出码: $?"
ls
python todo.py add 修复后新增
python todo.py list
ERROR: 存储文件损坏:todo.json 已改名备份,重新开始
(空)还没有待办事项
退出码: 0
__pycache__
demo_todo.json
todo.corrupted.json
todo.py
todo_store.py
已添加 #1: 修复后新增
[ ] 1 修复后新增

损坏文件没有被覆盖删除,而是整体改名为 todo.corrupted.json 留作线索(数据修复专家可能还能救回其中内容),程序从空列表重新开始,后续 add 照常工作。这就是原理部分说的”第二道保险”:原子写防的是”程序自己写坏”,损坏兜底防的是”外部把文件弄坏”,两道合起来,存储层的健壮性才算闭环。

第 5 步:为存储层写单元测试

上一篇的知识在这里交作业。测试目标明确限定在存储层:用临时目录隔离真实磁盘(每个测试独立 setUp/tearDown),把”编号递增、done 的 False 路径、损坏兜底”这些行为钉死。保存为 test_todo_store.py,与 todo_store.py 同目录:

# test_todo_store.py:只测存储层逻辑,用临时目录隔离真实磁盘
import logging
import tempfile
import unittest
from pathlib import Path
from todo_store import TaskStore

logging.disable(logging.CRITICAL)   # 测试期间静默日志,保证输出干净

class TestTaskStore(unittest.TestCase):
    def setUp(self):
        self.tmp = Path(tempfile.mkdtemp())
        self.store = TaskStore(self.tmp / 'todo.json')

    def tearDown(self):
        for p in self.tmp.iterdir():
            p.unlink()
        self.tmp.rmdir()

    def test_首次运行文件不存在(self):
        self.assertFalse(self.store.path.exists())   # 文件在首个任务之前不存在
        self.assertEqual(self.store.list_all(), [])  # 但 load 返回空列表而非报错

    def test_add落盘且编号递增(self):
        self.assertEqual(self.store.add('写单元测试'), 1)
        self.assertTrue(self.store.path.is_file())   # add 之后文件真实存在
        self.assertEqual(self.store.add('买菜'), 2)

    def test_done与remove的False路径(self):
        t1 = self.store.add('任务甲')
        self.store.add('任务乙')
        self.assertTrue(self.store.mark_done(t1))
        self.assertFalse(self.store.mark_done(999))  # 不存在的编号
        self.assertTrue(self.store.remove(t1))
        self.assertFalse(self.store.remove(999))
        tasks = self.store.list_all()
        self.assertEqual(len(tasks), 1)
        self.assertFalse(tasks[0]['done'])

    def test_新建实例能读回上次数据(self):
        self.store.add('重启后还在')
        other = TaskStore(self.tmp / 'todo.json')    # 模拟"下次运行"
        self.assertEqual(other.list_all()[0]['text'], '重启后还在')

    def test_损坏JSON兜底并可继续使用(self):
        (self.tmp / 'todo.json').write_text('{bad json', encoding='utf-8')
        self.assertEqual(self.store.list_all(), [])
        self.assertTrue((self.tmp / 'todo.corrupted.json').exists())
        self.store.add('损坏后仍可新增')
        self.assertEqual(len(self.store.list_all()), 1)

if __name__ == '__main__':
    unittest.main()

运行 python -m unittest -v test_todo_store:

test_add落盘且编号递增 (test_todo_store.TestTaskStore.test_add落盘且编号递增) ... ok
test_done与remove的False路径 (test_todo_store.TestTaskStore.test_done与remove的False路径) ... ok
test_损坏JSON兜底并可继续使用 (test_todo_store.TestTaskStore.test_损坏JSON兜底并可继续使用) ... ok
test_新建实例能读回上次数据 (test_todo_store.TestTaskStore.test_新建实例能读回上次数据) ... ok
test_首次运行文件不存在 (test_todo_store.TestTaskStore.test_首次运行文件不存在) ... ok

----------------------------------------------------------------------
Ran 5 tests in 0.092s

OK

五条测试对应五个行为契约:首次运行不炸、编号递增且落盘、不存在的编号返回 False(错误路径与成功路径都要测)、重启后数据仍在(持久化语义)、损坏文件兜底后仍可继续使用。logging.disable(logging.CRITICAL) 让损坏兜底那条测试不把 ERROR 日志打进测试报告,输出干净(第 4 篇的日志配置知识)。注意测试里创建 TaskStore 用的是临时路径而不是固定的 todo.json——测试永远不碰真实数据文件,这是测试隔离的底线。

到这里,一个”能用、耐摔、有测试”的命令行工具就完成了:四个子命令、一个 JSON 文件、47 行命令层加 50 行存储层、5 条测试。它现在的形态是一个”脚本集合”;下一篇文章将把它收拾成可安装的包(pyproject 打包),到时候存储层与命令层的分层就是包结构划分的现成依据。

易错点与陷阱

1. 首次运行直接读文件导致 FileNotFoundError。 新手版代码往往是 json.load(open(path)),工具第一次运行时文件根本不存在,直接崩溃。正确姿势是严格要求 load 的语义:先 exists() 判断,不存在返回空列表(本篇第 1 步);test_首次运行文件不存在 把这条行为固化成契约,防止未来重构时退化。

2. done/remove 不存在的编号静默通过。 如果 mark_done 找不到任务时什么都不做也不报错,用户敲错编号会以为”成功了”,数据却原地未动——静默失败比报错更危险。本篇的处理:返回 False,命令层写 ERROR 日志并退出码 1(第 6 篇约定),调用方一眼看出”#9 不存在”。测试里的 assertFalse(store.mark_done(999)) 正是防这个回归。

3. JSON 损坏时直接崩溃或直接覆盖。 不捕获 JSONDecodeError,程序每次启动都 traceback,用户无法使用;捕获后却直接 open(path, 'w') 重写,又把损坏现场销毁了。正确姿势(本篇第 4 步实测):报错日志 + 改名备份(.corrupted.json)+ 从空列表继续。备份是给”修复数据”留的后路。

4. 原子写时临时文件放错目录。 replace 的原子性依赖”源与目标在同一文件系统”。临时文件如果写在系统临时目录而目标在项目目录,跨卷 rename 可能抛 OSError 或退化成非原子复制。务必用 path.with_name(...) 让临时文件与目标同目录(本篇 save 的做法)。另外 Path.replace 的语义是覆盖式替换,Windows 与 POSIX 行为一致,可放心跨平台使用。

小结

本篇用纯标准库完成了一个可交付的待办命令行工具:存储层 TaskStore 负责 JSON 持久化(load 三路径:缺失/正常/损坏兜底)与原子写;命令层用 argparse 的 add_subparsers 声明 add/list/done/remove 四个子命令,按第 6 篇的退出码约定区分成功与业务错误;logging 把错误送进 stderr 与 stdout 输出分流。真实会话演示了增查改删全流程与损坏文件的自愈;5 条 unittest 把存储层的行为契约钉死。贯穿始终的两个工程原则值得带走:分层让 I/O 与逻辑隔离、测试才能真正落地;编号是身份不是序号,非法引用必须出声。下一篇将把这两个文件打包成可安装的 Python 包,第 10 篇测验则会对本项目的每个设计决策提问。

练习与思考题

  1. 给工具加第五个子命令 clear(清空全部任务):命令层注册 parser、存储层加 clear() 方法、测试补一条”清空后 list 为空且文件仍存在”。运行时注意:clear 不需要任何参数,用 sub.add_parser('clear') 即可。
  2. 手工在 todo.json 里把 "done": false 改成 "done": true(模拟人工编辑),运行 todo list,再说明:这个行为为什么能工作?如果把 done 字段设计成字符串 "True"/"False"(CSV 的教训),list 的判断代码会发生什么?
  3. 为”编号只增不减”写一条新测试:add 三条 → remove #1 → add 第四条,断言新增任务的 id 是 4 而不是 1;再说明如果把编号改成”删除后重排”,这条测试对应的契约会发生什么变化。
  4. 思考题:本工具的 JSON 文件如果被并发访问(同时开两个终端各跑一次 add),会发生什么?为什么”最后写入者胜”在这个量级可以接受,而到多用户场景就必须换数据库或加锁?(提示:想想 load 与 save 之间的时间窗;这是第 10 篇测验的隐藏考点。)