第 13 章 · 贯穿项目实战¶
理论学完了，现在用一个纯标准库项目把前 12 章串起来：logstats——一个分析服务器访问日志的命令行工具。它会用到 dataclass、生成器、collections、pathlib、argparse、datetime、re、类型提示、异常处理、pytest……几乎每一章的知识。
完整可运行代码在仓库的
project/目录，本章讲解设计与关键片段。
13.1 项目目标¶
logstats 读取访问日志，产出统计报告。假设日志每行格式：
即 <时间戳> <方法> <路径> <状态码> <耗时秒>。
支持子命令：
logstats stats access.log # 状态码分布、错误率、耗时统计
logstats top-slow access.log -n 5 # 最慢的 5 条请求
logstats stats *.log --json # 支持 glob 多文件、JSON 输出
13.2 数据模型：用 dataclass（第 4 章）¶
一条日志 = 一个不可变值对象。dataclass(frozen=True) 一行替代 Java 的 record + 构造器 + equals：
from dataclasses import dataclass
from datetime import datetime
@dataclass(frozen=True)
class LogEntry:
timestamp: datetime
method: str
path: str
status: int
duration: float
@property
def is_error(self) -> bool: # 派生属性，用 @property
return self.status >= 400
13.3 解析：正则 + 异常处理（第 5、9 章）¶
用 re 命名分组解析，失败行优雅跳过（不因一行脏数据崩溃整个程序）：
import re
LOG_RE = re.compile(
r"(?P<ts>\S+)\s+(?P<method>\w+)\s+(?P<path>\S+)"
r"\s+(?P<status>\d{3})\s+(?P<duration>[\d.]+)"
)
def parse_line(line: str) -> LogEntry | None:
m = LOG_RE.search(line)
if not m: # 无法解析 → 返回 None（第 8 章的 Optional 思想）
return None
try:
return LogEntry(
timestamp=datetime.fromisoformat(m["ts"]),
method=m["method"],
path=m["path"],
status=int(m["status"]),
duration=float(m["duration"]),
)
except (ValueError, KeyError):
return None # 任何字段转换失败都跳过该行
13.4 逐行读取：生成器（第 7 章）¶
日志可能很大（几个 GB）。用生成器逐行产出，不一次性载入内存：
from pathlib import Path
from collections.abc import Iterator
def iter_entries(path: Path) -> Iterator[LogEntry]:
with path.open(encoding="utf-8") as f: # with 自动关闭（第 7 章）
for line in f:
entry = parse_line(line)
if entry is not None:
yield entry # 惰性产出（第 7 章生成器）
支持多文件 / glob（pathlib 的 rglob/glob）：
def iter_all(patterns: list[str]) -> Iterator[LogEntry]:
for pattern in patterns:
for path in Path(".").glob(pattern): # 通配符展开
yield from iter_entries(path) # yield from 委托给子生成器
13.5 统计：collections + 推导式（第 7、9 章）¶
from collections import Counter
from statistics import mean, median
def summarize(entries: Iterator[LogEntry]) -> dict:
entries = list(entries) # 需要多次遍历就物化一次
durations = [e.duration for e in entries] # 推导式（第 7 章）
return {
"total": len(entries),
"status_codes": dict(Counter(e.status for e in entries)),
"error_rate": mean(e.is_error for e in entries) if entries else 0.0, # bool 求平均=比例；空列表兜底
"duration": {
"mean": mean(durations),
"median": median(durations),
},
}
13.6 CLI：argparse 子命令（第 9 章）¶
import argparse
import json
import sys
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(prog="logstats", description="Analyze access logs.")
sub = parser.add_subparsers(dest="cmd", required=True)
p_stats = sub.add_parser("stats", help="show summary statistics")
p_stats.add_argument("files", nargs="+", help="log files or glob patterns")
p_stats.add_argument("--json", action="store_true", help="output JSON")
p_slow = sub.add_parser("top-slow", help="show slowest requests")
p_slow.add_argument("files", nargs="+")
p_slow.add_argument("-n", type=int, default=5)
return parser
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
entries = list(iter_all(args.files)) # 统一物化
if args.cmd == "stats":
report = summarize(entries)
if getattr(args, "json", False):
print(json.dumps(report, ensure_ascii=False, indent=2, default=str))
else:
print(format_text(report)) # 自定义表格输出
elif args.cmd == "top-slow":
for e in sorted(entries, key=lambda x: x.duration, reverse=True)[: args.n]:
print(f"{e.duration:>7.3f}s {e.status} {e.method} {e.path}")
return 0
if __name__ == "__main__": # 第 6 章：入口
sys.exit(main())
13.7 测试：pytest（第 10 章）¶
# tests/test_parser.py
from logstats.parser import parse_line
def test_parse_valid():
e = parse_line("2026-06-16T10:00:01 GET /api/users 200 0.045")
assert e is not None
assert e.status == 200
assert e.is_error is False
def test_parse_garbage_returns_none():
assert parse_line("this is not a log line") is None
def test_parse_error_status():
e = parse_line("2026-06-16T10:00:02 POST /api/login 401 0.120")
assert e.is_error is True
用参数化覆盖更多边界（第 10 章）：
import pytest
@pytest.mark.parametrize("line, status", [
("2026-06-16T10:00:01 GET /a 200 0.1", 200),
("2026-06-16T10:00:02 GET /b 404 0.2", 404),
("2026-06-16T10:00:03 GET /c 500 0.3", 500),
])
def test_status_codes(line, status):
entry = parse_line(line)
assert entry is not None
assert entry.status == status
13.8 运行项目¶
cd project
uv sync # 安装项目（含开发依赖）
uv run logstats stats sample.log
uv run logstats stats '*.log' --json
uv run pytest # 跑测试
完整代码结构（project/）：
project/
├── pyproject.toml # [project.scripts] logstats = "logstats.cli:main"
├── sample.log
├── src/logstats/
│ ├── __init__.py
│ ├── models.py # LogEntry dataclass
│ ├── parser.py # parse_line / iter_entries
│ ├── stats.py # summarize / top-slow
│ └── cli.py # argparse + main()
└── tests/
├── test_parser.py
└── test_stats.py
13.9 这个项目串起了什么¶
| 章节知识点 | 在项目里体现 |
|---|---|
| 第 1 章 语法 | f-string 输出、集合、控制流 |
| 第 2 章 函数 | 类型注解签名、Iterator 返回 |
| 第 3 章 数据模型 | Counter/dict 统计 |
| 第 4 章 OOP | LogEntry dataclass + @property |
| 第 5 章 异常 | 解析容错、文件不存在处理 |
| 第 6 章 模块 | src/logstats/ 包结构、__main__ |
| 第 7 章 惯用法 | 生成器逐行读、推导式、with、yield from |
| 第 8 章 类型提示 | 全程注解 + LogEntry \| None |
| 第 9 章 标准库 | argparse/pathlib/re/json/statistics |
| 第 10 章 工程化 | pyproject.toml、pytest、src 布局 |
本章练习¶
练习 13.1
给 logstats 加一个子命令 paths，统计每个请求路径的出现次数，取 Top 10。
参考答案提示
复用 iter_entries + Counter(e.path for e in entries).most_common(10)。在 build_parser 加 sub.add_parser("paths")，main 加分支。
练习 13.2
给 summarize 增加耗时 P95（第 95 百分位）。提示：可用 statistics.quantiles 或手动排序。
参考答案提示
练习 13.3
让 logstats 支持读取 .gz 压缩日志（用 gzip 标准库），保持生成器逐行语义。
参考答案提示
在 iter_entries 里根据后缀选择打开方式：gzip.open(path, "rt", encoding="utf-8")（rt 文本模式），其余不变。
上一章：第 12 章 · 常见陷阱 ｜ 下一章：第 14 章 · 进阶路线图。