提升Python业务代码质量的9个实用工具库

分享9个提升Python开发效率的实用库,涵盖数据处理、类型检查、错误处理、性能分析等多个方面,助你写出更健壮、更易维护的代码。

原文标题:写业务代码必备:9 个被低估的 Python 高效工具库

原文作者:数据派THU

冷月清谈:

本文介绍了9个在业务代码开发中能够显著提升效率和代码质量的Python库。这些库包括:glom(用于处理嵌套数据)、boltons(弥补标准库的不足)、beartype(实现运行时类型检查)、result(强制显式错误处理)、whenever(避免时区错误)、pyinstrument(进行可读性强的性能分析)、dirty-equals(提供智能测试断言)、stamina(提供生产级重试逻辑)和pyfunctional(实现链式数据管道)。文章强调,选择合适的工具可以显著提高代码的可维护性和可读性。

怜星夜思:

1、beartype 看起来很棒,但过度使用运行时类型检查会不会影响性能?在哪些场景下应该谨慎使用它?
2、result 库引入了 Rust 的 Result 类型,这是否意味着 Python 也要开始走“类型安全”的路线了?你觉得这种趋势是好是坏?
3、文章提到了 pyinstrument 是一款优秀的性能分析工具,有没有其他类似的工具推荐?或者在使用性能分析工具时,有哪些容易忽略的坑?

原文内容

图片
来源:DeepHub IMBA
本文约3800字,建议阅读7分钟
本文介绍了 9 个实用 Python 库,高效解决业务开发各类痛点。


loguru、pydantic、httpx都是很好用的库,这篇文章整理的是另一类:多数开发者不知道它们存在,却在不少资深工程师的 requirements.txt 里出现。它们经过生产环境验证、持续维护,解决的都是写业务代码时反复遇到的具体问题。



glom:嵌套数据处理利器


几乎每个开发者都写过这样的代码:


 city = data.get("user", {}).get("profile", {}).get("address", {}).get("city", "Unknown")


能跑,但难看。一旦API多加一层嵌套,又得回来改这条链。


glom 把它变成声明式操作:


 from glom import glom, Coalesce  
data = {  
    "user": {  
        "profile": {  
            "address": {"city": "Lahore"},  
            "social": {"twitter": "@dev"}  
        }  
    }  
}  
# 简单的深层访问  
city = glom(data, "user.profile.address.city")  
# Returns: 'Lahore'  
# 当键不存在时提供回退值  
phone = glom(data, Coalesce("user.profile.phone", default="N/A"))  
# Returns: 'N/A'  
# 访问时同时进行转换 - 获取大写的Twitter用户名  
handle = glom(data, ("user.profile.social.twitter", str.upper))  
 # Returns: '@DEV'


glom 真正有意思的地方在于重构API响应——可以将整个输出模式定义为一个spec:


 from glom import glom, T  
spec = {  
    "city": "user.profile.address.city",  
    "handle": ("user.profile.social.twitter", T.upper()),  
    "name": "user.name"  
}  
result = glom(data | {"user": {**data["user"], "name": "Ali"}}, spec)  
 # {'city': 'Lahore', 'handle': '@DEV', 'name': 'Ali'}


一个 glom spec 可以替掉整个数据转换函数——那种30行的庞然大物。代码评审时看到这种写法,队友会印象深刻。


提示:glom 还提供 Assign 用于写入嵌套路径,不只是读取。它是一套完整的嵌套数据工具包。


boltons:弥补标准库缺失


Python的标准库有空白。总有些功能感觉理应存在却不存在。


boltons 填的就是这些空白。一组纯Python工具集,增补标准库的不足,涵盖 iterutils、strutils、timeutils 等模块。


 from boltons.iterutils import chunked, windowed, remap  
from boltons.strutils import slugify, camel2under  
# 将列表分成每3个一组  
data = list(range(10))  
print(list(chunked(data, 3)))  
# [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9]]  
# 在序列上进行滑动窗口操作  
print(list(windowed([1, 2, 3, 4, 5], 3)))  
# [(1, 2, 3), (2, 3, 4), (3, 4, 5)]  
# 深度清理嵌套结构:递归删除所有None值  
raw = {"a": 1, "b": None, "c": {"d": None, "e": 5}}  
cleaned = remap(raw, lambda p, k, v: v is not None)  
print(cleaned)  
# {'a': 1, 'c': {'e': 5}}  
# 开箱即用的字符串工具  
print(slugify("Hello World! This is Python."))  
# 'hello-world-this-is-python'  
print(camel2under("myVariableName"))  
 # 'my_variable_name'


单是 remap 函数就已经省去了无数次手写递归树遍历的工作。它遍历任意嵌套结构在一次遍历中完成过滤、转换或重构。


boltons 横跨20多个模块,包含超过230个工具。多数开发者发现它以后先用两个函数,六个月后又发现五个新的。


beartype:语言速度的运行时类型检查


Python类型提示默认只是装饰。写了 def add(a: int, b: int) -> int,传入字符串照样不会报错。运行时类型提示什么也不做。


beartype 让类型提示真正生效,而且速度快得超出预期。


from beartype import beartype  
from beartype.typing import List, Dict  
@beartype  
def process_users(users: List[Dict[str, str]]) -> List[str]:  
    return [u["name"] for u in users]  
# 这个可以正常工作  
print(process_users([{"name": "Alice"}, {"name": "Bob"}]))  
# 这个会立即抛出BeartypeException——而不是在下游产生一个静默的bug  
try:  
    process_users(["not", "a", "dict"])  
except Exception as e:  
     print(type(e).__name__, ":", e)


beartype 与其他同类工具的关键区别:O(1) 复杂度,常数时间。它不遍历整个列表逐一检查类型,而是采用概率采样策略和C级别的内省机制。面对百万元素的列表开销几乎为零。


 from beartype import beartype  
from beartype.typing import Annotated  
from beartype.vale import Is  
# 使用类型提示定义自定义验证器  
PositiveInt = Annotated[int, Is[lambda x: x > 0]]  
NonEmptyStr = Annotated[str, Is[lambda s: len(s) > 0]]  
@beartype  
def create_user(name: NonEmptyStr, age: PositiveInt) -> dict:  
    return {"name": name, "age": age}  
create_user("Alice", 25)       # Works  
create_user("", 25)            # Raises immediately  
 create_user("Alice", -1)       # Raises immediately


生产级别的输入验证,零行验证代码。只需要正确添加注解。


result:别再返回None


Python开发者都写过如下模式:


 def find_user(user_id):  
     user = db.query(user_id)  
     if user is None:  
         return None  # 调用方必须记得检查这个  
     return user


比如说如果忘了检查 if user is None,运行时就会冒出一个 AttributeError。


result 库将Rust的 Result 类型引入Python。函数要么返回 Ok(value),要么返回 Err(error),调用方被迫显式处理两种情况。


 from result import Ok, Err, Result, is_ok  
def divide(a: float, b: float) -> Result[float, str]:  
    if b == 0:  
        return Err("Division by zero is not allowed")  
    return Ok(a / b)  
def parse_age(value: str) -> Result[int, str]:  
    try:  
        age = int(value)  
        if age < 0:  
            return Err(f"Age cannot be negative: {age}")  
        return Ok(age)  
    except ValueError:  
        return Err(f"Cannot parse '{value}' as an integer")  
# 链式调用结果——只有在前一步成功时才执行下一步  
result = parse_age("25").and_then(lambda age: Ok(age * 365))  
print(result)  # Ok(9125)  
result = parse_age("abc")  
print(result)  # Err("Cannot parse 'abc' as an integer")  
# 使用回退值安全地解包  
age_in_days = result.unwrap_or(0)  
 print(age_in_days)  # 0


显式的错误处理成为强制要求,不再有被静默吞掉的 None 值。函数签名如实反映了可能出错的情况。


whenever:日期和时间的正确做法


基本上每个Python项目最终都会遇到时区bug。用了 datetime.now(),忘了它是简单的时间,把它和一个带时区的datetime混到一起,调度器会在错误的时间触发。


whenever 在类型层面就杜绝了时区错误,是一个现代的日期时间库:


from whenever import ZonedDateTime, hours  
# 从一开始就使用明确的、带时区的日期时间  
meeting = ZonedDateTime(2025, 3, 17, 14, 30, tz="America/New_York")  
# 正确地转换到任何时区  
print(meeting.as_tz("Asia/Karachi"))  
# ZonedDateTime(2025-03-17 23:30:00+05:00[Asia/Karachi])  
# 遵守夏令时(DST)的时间运算  
next_week = meeting + hours(168)  
print(next_week)  
# 跨时区的瞬时比较  
lahore_time = ZonedDateTime(2025, 3, 17, 23, 30, tz="Asia/Karachi")  
print(meeting == lahore_time)  # True - 同一时刻,不同时区


和 pendulum 比,whenever 有Rust实现的性能优势,并且会将DST歧义时间作为错误抛出而不是静默猜测。后面这一点在生产环境中坑过一次。一次就够了。


pyinstrument:像资深工程师一样做性能分析


代码慢的时候,多数开发者第一反应是 cProfile。盯着300行输出看了半天,然后放弃。


pyinstrument 是统计采样型分析器,输出对人类可读。它直接告诉时间花在了哪里,没有多余的噪音:


from pyinstrument import Profiler  
import time  
def slow_function():  
    time.sleep(0.1)  
    return sum(range(1_000_000))  
def fast_function():  
    return 42  
def main():  
    for _ in range(5):  
        slow_function()  
    for _ in range(1000):  
        fast_function()  
profiler = Profiler()  
profiler.start()  
main()  
profiler.stop()  
profiler.print()  


输出如下——干净、可读、立即可操作:  


  _     ._   __/__   _ _  _  _ _/_   Recorded: 14:23:01  Samples:  52  
 /_//_/// /_\ / //_// / //_'/ //     Duration: 0.521     CPU time: 0.498  
/   _/                      v4.6.2  
0.521 main  <your_script.py>:14  
└── 0.521 slow_function  <your_script.py>:5  
    └── 0.521 sleep  <built-in>


也可以从命令行直接运行,无需修改任何代码:


pyinstrument your_script.py


Dropbox和Instagram的工程师在生产环境中使用过统计分析器,因为它们引入的开销不到1%——而确定性分析器可能让代码慢10到100倍。


dirty-equals:不说谎的测试断言


测试复杂的API响应是件苦差事。响应中有每次运行都会变化的时间戳,有无法预测的UUID。结果要么什么都不测,要么写40行断言体操。


dirty-equals 的解法是提供一组智能比较对象:


from dirty_equals import IsDatetime, IsUUID, IsPositiveInt, IsStr, Contains  
response = {  
    "id": "3f7a9c21-4d8e-4b12-a765-0f3d8e1c9b2a",  
    "created_at": "2025-03-17T14:30:00Z",  
    "user_count": 42,  
    "message": "Welcome back, Alice!",  
    "tags": ["python", "backend", "api"]  
}  
# 断言结构而不硬编码易变的值  
assert response == {  
    "id": IsUUID,  
    "created_at": IsStr(regex=r'\d{4}-\d{2}-\d{2}T.*'),  
    "user_count": IsPositiveInt,  
    "message": IsStr & Contains("Alice"),  
    "tags": Contains("python")  
}  
print("All assertions passed!")


测试从脆弱的快照比较转变为意图驱动的断言——验证的是数据的含义,而非某个精确毫秒的快照值。


提示:dirty-equals 还包含 IsApprox(浮点数比较)、IsJson、IsList(length=5) 等,可与 pytest 及任何断言库配合使用。


stamina:生产级重试逻辑,兼容异步


tenacity 可能大家都比较熟悉,stamina 可以理解为 tenacity 经过深思熟虑的生产环境版本。


默认配置即合理——带jitter的指数退避、按异常类型可配置;更关键的是它与 structlog 和 prometheus 集成,重试尝试自动具备可观测性。


import stamina  
import httpx  
# 同步版本  
@stamina.retry(on=httpx.HTTPError, attempts=5)  
def fetch_sync(url: str) -> dict:  
    r = httpx.get(url, timeout=5.0)  
    r.raise_for_status()  
    return r.json()  
# 异步版本——直接可用  
@stamina.retry(on=httpx.HTTPError, attempts=5, timeout=30.0)  
async def fetch_async(url: str) -> dict:  
    async with httpx.AsyncClient() as client:  
        r = await client.get(url, timeout=5.0)  
        r.raise_for_status()  
        return r.json()  
# 添加监控:stamina会自动将重试次数  
# 发送到structlog和prometheus(如果它们在你的项目中)  
stamina.instrument(logger=True)


jitter防止所有服务同时重试时产生惊群效应的随机小延迟,所以会默认开启。tenacity 需要手动添加,stamina 直接作为默认值。


pyfunctional:链式数据管道


面向对象的Python在需要对一个列表连续执行8个操作时就开始费力了。要么写一个嵌套极深的单行表达式,要么创建8个无人问津的中间变量。


pyfunctional 提供惰性的、可链式调用的数据管道:


from functional import seq  
employees = [  
    {"name": "Alice", "dept": "Engineering", "salary": 120000},  
    {"name": "Bob",   "dept": "Marketing",   "salary": 85000},  
    {"name": "Carol", "dept": "Engineering", "salary": 135000},  
    {"name": "Dave",  "dept": "Marketing",   "salary": 92000},  
    {"name": "Eve",   "dept": "Engineering", "salary": 110000},  
]  
# 获取工程部门中薪资超过10万的平均薪资  
result = (  
    seq(employees)  
    .filter(lambda e: e["dept"] == "Engineering")  
    .filter(lambda e: e["salary"] > 100000)  
    .map(lambda e: e["salary"])  
    .average()  
)  
print(f"Avg Engineering salary (>100k): ${result:,.0f}")  
# Avg Engineering salary (>100k): $121,667  
# 按部门分组,获取每个部门的最高薪资  
by_dept = (  
    seq(employees)  
    .group_by(lambda e: e["dept"])  
    .map(lambda kv: (kv[0], max(e["salary"] for e in kv[1])))  
    .to_dict()  
)  
print(by_dept)  
# {'Engineering': 135000, 'Marketing': 92000}


惰性求值,在调用 .average() 或 .to_list() 等终端操作之前不会执行任何计算。读起来接近自然语言,并且原生支持CSV、JSON和数据库行。


和 for 循环加三个中间列表的写法对比一下,优点一眼就能看出来。


总结


这9个库覆盖了日常开发中几个反复出现的痛点:嵌套数据访问、标准库功能缺失、运行时类型安全、错误处理模式、时区陷阱、性能分析、测试断言、重试机制和数据管道。它们的共同特点是API设计克制、默认行为合理,拿来就能用在生产环境中。


区分业余代码和生产级代码的往往不是算法或架构,而是对这些细节层面工具的选择。下次新建项目时,把其中几个加进 requirements.txt 试试,代码的可维护性和可读性会有明显变化。


by Abdur Rahman


编辑:于腾凯

校对:林亦霖



关于我们

数据派THU作为数据科学类公众号,背靠清华大学大数据研究中心,分享前沿数据科学与大数据技术创新研究动态、持续传播数据科学知识,努力建设数据人才聚集平台、打造中国大数据最强集团军。




新浪微博:@数据派THU

微信视频号:数据派THU

今日头条:数据派THU

Result也好,Beartype也好,都体现了Python社区对代码质量和可维护性的重视。虽然Python是动态类型语言,但并不意味着可以放任类型错误不管。引入Result,可以迫使开发者显式地处理错误,减少程序崩溃的可能性。至于是不是要走“类型安全”的路线,我觉得这是一个渐进的过程。Python会吸收其他语言的优点,但不会完全抛弃自己的灵活性。

分享一个我踩过的坑:在分析多线程程序时,要注意线程切换的开销。有些分析工具可能无法准确地反映线程切换的时间,导致分析结果不准确。另外,要避免在生产环境进行性能分析,以免影响线上服务。如果需要在生产环境进行分析,可以使用统计采样型分析器,比如 pyinstrument,它们引入的开销比较小。

我觉得没必要过度解读。Result 只是一个工具,用不用取决于具体的场景。有些项目追求快速迭代,可以牺牲一些类型安全;有些项目对稳定性要求很高,就可以选择 Result。Python 的哲学是“There should be one-- and preferably only one --obvious way to do it”,但并不排斥其他的可能性。Result 提供了一种新的选择,仅此而已。

我个人非常支持这种趋势。Python的灵活性是把双刃剑,用得好可以提高效率,用不好就容易挖坑。Result 这种机制,相当于在动态类型的基础上增加了一层约束,让代码更健壮。而且,显式错误处理也有助于提高代码的可读性和可理解性。当然,任何改变都会有争议,但只要能解决实际问题,就是值得尝试的。