一个 Python 项目里,如果到处都是 os.path.join、临时字典、状态字符串和大段 try...finally,我一般不急着抽工具类。这种代码看着只是多写几行,时间长了才麻烦:路径处理有三套写法,数据分组各写各的,临时文件删不干净,状态值还经常拼错。
Python 标准库里已经有现成的。下面这 8 个,我在日志处理、数据导入、接口兜底和批处理脚本里用得比较多。
1. pathlib:别再手拼文件路径
看到下面这种代码,我第一反应就是换掉:
import oslog_file = os.path.join(root_dir, "logs", date_str, "error.log")if os.path.exists(log_file) and os.path.isfile(log_file):with open(log_file, encoding="utf-8") as file: content = file.read()
用 pathlib 后,路径就是一个对象,查找、判断和读取都放在一起:
from pathlib import Pathdeffind_error_logs(work_dir: str): log_dir = Path(work_dir) / "logs"for log_file in log_dir.rglob("*.log"):with log_file.open(encoding="utf-8", errors="ignore") as stream:if any("ERROR"in line for line in stream):yield log_file
批量扫日志、整理上传目录、归档文件时,用它能少写不少路径判断。
2. dataclasses:数据对象没必要手写一堆模板
接口返回值、导入记录、任务参数这类对象,经常只有字段,没有复杂行为。继续手写构造函数和打印方法,纯属给自己加活。
from dataclasses import dataclassfrom decimal import Decimal@dataclass(slots=True)classImportOrder: order_no: str shop_code: str amount: Decimal source: str = "excel" passed: bool = True
dataclass 会自动生成初始化、比较和字符串展示代码。
我一般还会加上 slots=True。这种批量导入对象可能一次创建几万条,没必要让每个实例都带一个随意扩展的 __dict__。
3. defaultdict:分组代码别反复判断键存不存在
普通字典做数据分组,经常写成这样:
shop_orders = {}for order in orders:if order.shop_code notin shop_orders: shop_orders[order.shop_code] = [] shop_orders[order.shop_code].append(order)
这种判断没什么技术含量,但项目里特别多。
from collections import defaultdictshop_orders = defaultdict(list)for order in orders: shop_orders[order.shop_code].append(order)
除了 list,还可以传 set、int,甚至自定义工厂函数。做字段聚合、错误收集和批量分类时很顺手。
4. Counter:统计次数别自己维护加一逻辑
统计接口错误码时,有些代码会先判断键,再决定赋值为 1 还是加 1。能跑,但看着费劲。
from collections import Countererror_codes = Counter( record["error_code"]for record in request_logsif record.get("error_code"))print(error_codes.most_common(5))
Counter 不只是计数,还能直接拿出现次数最高的数据。
排查线上问题时,我经常先把当天失败日志里的错误码扫一遍。几十种异常混在一起时,先看最高频的,比从第一条日志往下翻靠谱。
5. itertools:多批数据不要一层层拼列表
比如订单来源有三处:正常查询、失败重试和人工补单。很多代码会先创建一个大列表,再连续调用三次 extend。
其实没必要把数据提前复制一遍。
from itertools import chainall_order_ids = chain( query_pending_orders(), query_retry_orders(), load_manual_orders(),)for order_id in all_order_ids: dispatch_order(order_id)
chain 是惰性读取,前一批处理完才会继续取下一批。
数据量大时,这一点比少写几行更重要。为了代码“看起来统一”,先把几十万条数据塞进一个新列表,我不太喜欢这种写法。
6. functools.lru_cache:稳定查询别重复执行
地区编码、商品分类、权限规则这类数据,在一次任务里可能被反复查询。
自己维护缓存字典,通常还得处理键判断和容量问题。
from functools import lru_cache@lru_cache(maxsize=512)defload_region_name(region_code: str) -> str: row = region_repository.find_by_code(region_code)if row isNone:return"UNKNOWN"return row.name
后面的调用不需要知道缓存存在:
region_name = load_region_name("310000")
不过这个东西不能乱套。库存、余额、任务状态这种实时变化的数据,我不会缓存。缓存旧了,比多查一次更麻烦。
7. contextlib.ExitStack:动态打开多个文件时很好用
批量校验完成后,可能要把成功、失败和待确认数据分别写入不同文件。
文件数量固定时,写多个 with 还能忍。文件由运行时决定,就容易堆出一排打开和关闭代码。
import csvfrom contextlib import ExitStackfrom pathlib import Pathdefexport_rows(output_dir: str, grouped_rows: dict): target = Path(output_dir) target.mkdir(parents=True, exist_ok=True)with ExitStack() as stack: writers = {}for category in grouped_rows: stream = stack.enter_context( (target / f"{category}.csv").open("w", encoding="utf-8-sig", newline="", ) ) writers[category] = csv.writer(stream)for category, rows in grouped_rows.items(): writers[category].writerows(rows)
中途抛异常也不用自己逐个关文件,ExitStack 会按顺序处理。
这种资源清理代码,能交给标准库就别自己逞强。
8. enum:业务状态别散落成字符串
订单状态如果一直用 "WAIT_PAY"、"PAID" 这种字符串,早晚会出现大小写不一致或者拼写错误。
from enum import StrEnumclassOrderStatus(StrEnum): WAIT_PAY = "WAIT_PAY" PAID = "PAID" CLOSED = "CLOSED"
业务判断也会清楚一些:
defcan_close(status: OrderStatus) -> bool:return status in { OrderStatus.WAIT_PAY, OrderStatus.PAID, }
这里我不建议为了“统一管理”把所有常量都塞进枚举。真正有固定取值范围、会参与业务判断的状态才适合放进去。
标准库减少重复代码,靠的不是语法炫技。
路径处理交给 pathlib,数据对象交给 dataclasses,分组和统计交给 collections,资源释放交给 contextlib。这些地方自己再造一套工具层,开始可能觉得灵活,半年后通常只剩下一堆没人敢删的封装。
代码少一点,分支少一点,出问题时也少翻几层。