Python 工程化

本章目标：把一段能跑的 Python 脚本拆成有目录、有配置、有日志、有测试的小项目，并能在命令行稳定执行。 前置知识：Git 与 Python 基础 预计用时：120 分钟

会写 for 循环和 pandas，只能算“能写脚本”。数据开发每天要交付的是一批会反复运行、别人也要接手的任务。本章把“脚本”升级成“工程”：目录清楚、依赖可复现、出错能定位、改完敢上线。

1. 为什么需要工程化

先看一个常见脚本：

import pandas as pd

df = pd.read_csv("C:/Users/me/Desktop/orders.csv")
df = df[df["status"] != "canceled"]
print(df.groupby("city")["amount"].sum())

它在写的人电脑上能跑，换一台机器就可能出四个问题：

• 路径写死，换目录就找不到文件。
• 没有说明需要 pandas，别人装不上。
• 出错只抛一大段 traceback，不知道是哪一行数据坏了。
• 没有测试，改一行代码不知道有没有算错。

工程化不是为了“显得专业”，而是为了让任务可复现、可定位、可验证。这三件事直接决定你能不能把任务交给调度系统每天跑。

2. 一个数据任务的最小结构

jobs/
├── __init__.py
├── etl/
│   ├── __init__.py
│   ├── cli.py          # 命令行入口
│   ├── config.py       # 路径和参数
│   └── transform.py    # 纯计算逻辑
└── tests/
    └── test_transform.py
requirements.txt
pyproject.toml

分工原则：

把“读参数”和“算数据”分开，测试才能只测计算，不用模拟命令行。

3. 动手：写一个可复现的汇总任务

下面所有命令都在项目根目录执行。先建目录和空文件：

mkdir -p jobs/etl jobs/tests
touch jobs/__init__.py jobs/etl/__init__.py

3.1 纯计算逻辑

新建 jobs/etl/transform.py：

from __future__ import annotations

import csv
import logging
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from pathlib import Path

log = logging.getLogger(__name__)


class DataError(ValueError):
    """输入数据不符合约定时抛出，带上行列信息。"""


@dataclass(frozen=True)
class Order:
    order_id: str
    status: str
    city: str
    amount: Decimal


def read_orders(path: Path) -> list[Order]:
    rows: list[Order] = []
    with path.open(newline="", encoding="utf-8") as handle:
        reader = csv.DictReader(handle)
        for lineno, raw in enumerate(reader, start=2):
            try:
                amount = Decimal(raw["amount"])
            except (KeyError, InvalidOperation) as exc:
                raise DataError(f"{path}:{lineno} amount 不是合法数字") from exc
            if amount < 0:
                raise DataError(f"{path}:{lineno} amount 不能为负")
            rows.append(Order(raw["order_id"], raw["status"], raw["city"], amount))
    return rows


def gmv_by_city(orders: list[Order]) -> dict[str, Decimal]:
    totals: dict[str, Decimal] = {}
    for order in orders:
        if order.status in {"canceled", "unavailable"}:
            continue
        totals[order.city] = totals.get(order.city, Decimal("0")) + order.amount
    return dict(sorted(totals.items(), key=lambda item: item[1], reverse=True))

这里有两个值得注意的选择：

• 金额用 Decimal，不用 float。财务数字用二进制浮点会出现 0.1 + 0.2 != 0.3 这类误差。
• read_orders 出错时抛 DataError 并带上行号，排错时不用猜是哪条数据。

3.2 命令行入口

新建 jobs/etl/cli.py：

from __future__ import annotations

import argparse
import logging
from pathlib import Path

from .transform import DataError, gmv_by_city, read_orders

log = logging.getLogger(__name__)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="按城市汇总有效订单金额")
    parser.add_argument(
        "--input",
        type=Path,
        default=Path("data/sample/orders_demo.csv"),
        help="订单 CSV 路径",
    )
    parser.add_argument(
        "--log-level",
        default="INFO",
        choices=["DEBUG", "INFO", "WARNING", "ERROR"],
    )
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    logging.basicConfig(
        level=args.log_level,
        format="%(asctime)s %(levelname)s %(name)s %(message)s",
    )

    try:
        orders = read_orders(args.input)
    except FileNotFoundError:
        log.error("输入文件不存在：%s", args.input)
        return 2
    except DataError as exc:
        log.error("数据校验失败：%s", exc)
        return 3

    log.info("读取 %d 行，开始汇总", len(orders))
    for city, amount in gmv_by_city(orders).items():
        print(f"{city}\t{amount:.2f}")
    return 0


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

运行：

python -m jobs.etl.cli --input data/sample/orders_demo.csv

输出按金额从高到低：

北京  455.00
上海  405.00
广州  319.00
深圳  180.00
杭州  150.00

把日志级别调成 DEBUG 不会改变结果，但会多打印过程日志。生产任务通常把日志写到文件，再被采集系统收走。

3.3 测试计算逻辑

新建 jobs/tests/test_transform.py：

import unittest
from decimal import Decimal

from jobs.etl.transform import Order, gmv_by_city


class GmvByCityTest(unittest.TestCase):
    def test_skip_canceled_and_unavailable(self):
        orders = [
            Order("1", "paid", "北京", Decimal("10")),
            Order("2", "canceled", "北京", Decimal("99")),
            Order("3", "unavailable", "北京", Decimal("88")),
        ]
        self.assertEqual(gmv_by_city(orders), {"北京": Decimal("10")})

    def test_sort_desc(self):
        orders = [
            Order("1", "paid", "A", Decimal("1")),
            Order("2", "paid", "B", Decimal("2")),
        ]
        self.assertEqual(list(gmv_by_city(orders)), ["B", "A"])


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

运行：

python -m unittest discover -s jobs/tests -t . -v

应该看到 2 个测试通过。测试不读真实 CSV，改文件格式不会误伤计算逻辑；读文件的部分由 read_orders 自己负责。

3.4 依赖与代码规范

新建 requirements.txt：

duckdb>=1.0,<2.0
ruff>=0.6

新建 pyproject.toml：

[tool.ruff]
line-length = 100
target-version = "py311"

安装依赖并检查代码：

python -m pip install -r requirements.txt
python -m ruff check jobs

requirements.txt 记录“需要什么”，虚拟环境负责“装在哪”。两者不要混：不要把 .venv/ 提交到 Git。

4. 配置、日志与失败处理

4.1 配置从环境变量来

任务经常要在开发、测试、生产三套环境跑。路径和连接信息不要写死在代码里：

import os
from pathlib import Path

INPUT_PATH = Path(os.getenv("ETL_INPUT", "data/sample/orders_demo.csv"))
LOG_LEVEL = os.getenv("ETL_LOG_LEVEL", "INFO")

本地不设置环境变量就用默认值；调度系统运行时注入生产值。密钥不要放进代码和 requirements.txt，用环境变量或密钥管理服务。

4.2 日志不是 print

print 只适合看结果。日志要能回答：什么时候、哪个任务、哪一步、处理了多少行、失败原因是什么。

log.info("读取 %d 行", len(orders))
log.warning("跳过 %d 行状态为 canceled 的记录", canceled)
log.exception("写结果失败")

日志级别约定：

• DEBUG：调试细节，生产通常关闭。
• INFO：任务开始、结束、处理行数。
• WARNING：可以继续，但需要关注，例如脏数据被跳过。
• ERROR：任务失败，需要人工处理。

4.3 失败要返回非零退出码

调度系统靠退出码判断成功失败。捕获异常后打印日志，再 return 2 或 raise SystemExit(2)；不要让任务“打印了错误但退出码还是 0”。

注意：不要用 except Exception: pass 吞掉异常。那会让任务看起来成功，数据却没写进去。

5. 常见坑

5.1 路径依赖当前目录

open("data/orders.csv") 只在项目根目录执行时有效。调度系统的工作目录可能不同。用 Path(__file__).resolve().parent 定位代码目录，或者把输入路径作为参数传入。

5.2 把密钥写进代码

数据库密码、API Key 写进 .py 文件，提交到 Git 就等于公开。用 .env 本地加载、环境变量注入生产，.env 加进 .gitignore。

5.3 测试只测“能跑”

只断言“没抛异常”没有意义。测试要断言具体结果，例如金额等于 455.00、排序是 B 在 A 前。

5.4 一次任务做太多事

一个脚本里读数据、清洗、聚合、写库、发邮件、更新看板，失败时根本不知道停在哪。拆成多个步骤，每步有输入输出和日志，调度器才能单独重跑。

5.5 忽略时区和编码

CSV 可能带 BOM，Windows 默认编码也不一定是 UTF-8。读文件时显式指定 encoding="utf-8"；时间字段统一存 UTC 或明确业务时区，写进字段注释。

6. 在线练习

这三题不是本章主考点，用来保持 SQL 手感。数据开发岗位不会只考 Python，也会要求你能用 SQL 验证脚本结果。

自测：

• 能不看资料写出 argparse 的必填参数和默认值。
• 能解释 Decimal 和 float 在金额计算上的区别。

7. 面试延伸

面试题：为什么数据任务要做成可配置、可重跑？

是什么：可配置指路径、日期、并发等参数从外部传入；可重跑指同一批数据跑多次结果一致。

解决什么问题：调度失败要补数，补数不能改代码；不同环境不能各维护一份脚本；出了问题要能定位到具体参数。

我的实操边界：本章的脚本在本地跑通，未接入真实调度集群；生产上的密钥管理和多环境发布还需要平台支持。

面试题：日志和 print 的区别？

是什么：日志有级别、时间、模块名，可以输出到文件并被采集；print 只写标准输出。

解决什么问题：线上任务没人盯着屏幕，只能靠日志回溯。日志级别还能控制生产环境的信息量。

我的实操边界：本地用标准库 logging，没有接 ELK/Loki 这类日志平台。

面试题：怎么保证脚本换台机器还能跑？

是什么：固定 Python 版本、锁定依赖、显式路径和编码，并把启动命令写进 README。

解决什么问题：避免“在我电脑上能跑”。容器化可以进一步固化系统依赖，见 容器化部署。

我的实操边界：本章用 venv 和 requirements 解决 Python 依赖，未覆盖系统级依赖。

8. 本章验收

• 能画出 cli.py、config.py、transform.py、tests/ 的职责边界。
• 能独立运行 python -m jobs.etl.cli --input data/sample/orders_demo.csv 并得到 5 个城市的结果。
• 能写出至少 2 个单元测试，并让 unittest 全部通过。
• 能说明为什么金额用 Decimal，以及 float 会带来什么风险。
• 能说出日志四个级别各自适合记录什么。
• 能把一个写死路径的脚本改成从参数或环境变量读取输入。