社区所有版块导航
Python
python开源   Django   Python   DjangoApp   pycharm  
DATA
docker   Elasticsearch  
aigc
aigc   chatgpt  
WEB开发
linux   MongoDB   Redis   DATABASE   NGINX   其他Web框架   web工具   zookeeper   tornado   NoSql   Bootstrap   js   peewee   Git   bottle   IE   MQ   Jquery  
机器学习
机器学习算法  
Python88.com
反馈   公告   社区推广  
产品
短视频  
印度
印度  
Py学习  »  Python

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

AI思想会 • 3 月前 • 176 次点击  
【前言】AI 正以前所未有的速度发展,新的机遇不断涌现,如果你希望:与技术专家、产品经理和创业者深度交流,一起探索 AI如何改变各行各业。欢迎在文末扫二维码,加入「AI思想会」交流群,和一群志同道合的伙伴共同学习、思考、创造!

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([12345], 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 ListDict  @beartype  def process_users(users: List[Dict[strstr]]) -> 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[floatstr]:      if b == 0:          return Err("Division by zero is not allowed")      return Ok(a / b)  def parse_age(value: str) -> Result[intstr]:      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(20253171430, 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(20253172330, 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  :14  └── 0.521 slow_function  :5      └── 0.521 sleep  


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


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

本文系学术转载,如有侵权,请联系后台删文
(👇 长按识别二维码)
图片
转自:DeepHub IMBA

Python社区是高质量的Python/Django开发社区
本文地址:http://www.python88.com/topic/194701