Administrator
Administrator
发布于 2026-10-31 / 0 阅读
0
0

Python 从入门到初级(13):模块与包:把订单逻辑拆成可复用组件

本篇是《Python 从入门到初级》第 13 / 20 篇。示例使用 Python 3.12 兼容语法,完整目录结构与数据见配套源码包。

前面的 CSV 脚本已经能生成报告,但如果另一个程序也要统计订单,不应该再复制整段代码。本篇把“读数据”和“做汇总”拆成模块,为后面的数据库、命令行和测试复用。

读取与统计封装成模块,由不同入口导入调用

1. 模块、包和入口分别是什么

一个 .py 文件可以是模块;包把相关模块组织到目录中;入口负责接收用户命令并安排调用。

这里使用传统的带 __init__.py 的包结构:

examples/
├── lesson13.py
├── data/orders.csv
└── ops_tools/
    ├── __init__.py
    └── orders.py

__init__.py 本例可以只放一句包说明。不要把自己的文件命名为 csv.py、json.py 或 pathlib.py,否则可能遮住要导入的标准库模块。

2. 先给可复用模块写清输入约定

订单模块接受含必需表头的 CSV,要求编号不重复、分类非空、金额为非负整数分,状态只能为 paid、pending、cancelled。空数据表允许,但表头必须存在。

相较于第 11 篇,它不再对错误记录“算了再说”,而是在读取边界拒绝不符合约定的数据,让调用者决定提示、停止还是处理其他文件。

保存 ops_tools/orders.py:

import csv
import json
from pathlib import Path

REQUIRED_FIELDS = {"order_id", "category", "amount_cents", "status"}
STATUSES = {"paid", "pending", "cancelled"}


def read_orders(path):
    orders = []
    seen_ids = set()
    with Path(path).open(newline="", encoding="utf-8-sig") as handle:
        reader = csv.DictReader(handle)
        if reader.fieldnames is None or not REQUIRED_FIELDS.issubset(reader.fieldnames):
            raise ValueError("CSV 缺少必需表头")
        for row in reader:
            line_number = reader.line_num
            if None in row or any(row.get(field) is None for field in REQUIRED_FIELDS):
                raise ValueError(f"第 {line_number} 行字段数量不匹配")
            order_id = row["order_id"].strip()
            category = row["category"].strip()
            status = row["status"].strip()
            if not order_id or not category:
                raise ValueError(f"第 {line_number} 行编号或分类为空")
            if order_id in seen_ids:
                raise ValueError(f"第 {line_number} 行订单编号重复")
            if status not in STATUSES:
                raise ValueError(f"第 {line_number} 行状态不受支持")
            try:
                amount = int(row["amount_cents"])
            except ValueError as exc:
                raise ValueError(f"第 {line_number} 行金额必须是整数分") from exc
            if amount < 0:
                raise ValueError(f"第 {line_number} 行金额不能为负数")
            seen_ids.add(order_id)
            orders.append({"order_id": order_id, "category": category,
                           "amount_cents": amount, "status": status})
    return orders


def summarize(orders):
    paid = [order for order in orders if order["status"] == "paid"]
    categories = {}
    for order in paid:
        name = order["category"]
        categories[name] = categories.get(name, 0) + order["amount_cents"]
    return {
        "total_orders": len(orders),
        "paid_orders": len(paid),
        "paid_amount_cents": sum(order["amount_cents"] for order in paid),
        "by_category_cents": dict(sorted(categories.items())),
    }


def save_json(summary, path):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("x", encoding="utf-8") as handle:
        json.dump(summary, handle, ensure_ascii=False, indent=2)
        handle.write("\n")

3. 三个函数为什么分开

read_orders() 负责文件和数据验证,返回清洗过的记录列表;summarize() 负责对符合约定的记录计算汇总;save_json() 负责输出。

这样测试计算时不必总读真实文件,换成数据库或 HTTP 入口时,也有清晰的复用边界。summarize() 仍要求调用方提供符合字段约定的记录,并不是接受任何 Python 对象的万能转换器。

save_json 使用 x 模式,输出文件存在时拒绝覆盖。失败写入不具备自动回滚或原子替换能力,后面的工具会明确报告错误;这仍是一个入门实现,不是事务式文件存储。

4. 编写一个很薄的入口

保存并运行 lesson13.py:

from pathlib import Path
from ops_tools.orders import read_orders, summarize


def main():
    base = Path(__file__).resolve().parent
    orders = read_orders(base / "data" / "orders.csv")
    summary = summarize(orders)
    print("订单数:", summary["total_orders"])
    print("已支付金额(分):", summary["paid_amount_cents"])


if __name__ == "__main__":
    main()

预期订单数为 5,已支付金额为 38300 分。入口知道数据在哪,模块知道怎样处理数据;两者没有相互抢占职责。

from ops_tools.orders import ... 从包中导入所需名字。Python 的模块加载机制见 官方 Modules 教程。运行时应保留目录结构,不要只把入口文件单独复制走。

5. if __name__ 的作用

当文件作为主程序运行时,__name__ 为 "__main__";作为模块导入时,它通常是模块名。把启动动作放在这个判断下面,可以让“导入函数”和“执行工具”分开。

模块顶层的其他语句仍会在首次导入执行,所以不要在顶层无条件调用接口、复制文件或删除数据。本例模块只定义常量和函数。

6. 怎样确认导入的是哪份代码

在项目根目录可以查看模块位置:

python -c "import ops_tools.orders as m; print(m.__file__)"

遇到 ModuleNotFoundError,先查当前项目目录、解释器和包结构。不要立即把各种目录硬塞进 sys.path,或把项目文件复制到系统安装目录。

练习:给数据增加重复编号,预期读入时被明确拒绝;恢复原始文件,再导入 summarize 计算空列表,预期数量和金额为 0。它们检验的是模块的输入约定,而不只是 import 是否成功。


上一篇:时间、正则与 Counter:统计日志和异常 · 系列目录 · 下一篇:类与对象:建立带行为的设备档案


评论