Python 3.15：未登上头条的功能

# Python 3.15：那些未登头条的特性 来源：https://blog.changs.co.uk/python-315-features-that-didnt-make-the-headlines.html 又是一年一度 Python 新版本发布的时节。随着 Python 3.15.0b1 (https://docs.python.org/3.15/whatsnew/3.15.html) 功能冻结，我们已经知道今年晚些时候 Python 将会带来哪些新特性。有许多重磅功能即将到来，包括惰性导入 (https://docs.python.org/3.15/whatsnew/3.15.html#whatsnew315-lazy-imports) 和太赫兹分析器 (https://docs.python.org/3.15/whatsnew/3.15.html#whatsnew315-sampling-profiler)（我之前曾 (https://blog.changs.co.uk/benchmarking-free-threading-performance-with-tachyon.html) 介绍过后者）。 去年，我非常享受探索 (https://blog.changs.co.uk/python-314-3-smaller-features.html) Python 3.14 中那些较小特性的过程。我发现其中许多特性与大 PEP 同样有趣，理应得到更多关注。今年情况亦如此。 ## Asyncio 任务组取消 这个版本的 Asyncio 改动不多。主要新增的功能是能够优雅地取消 `TaskGroup`。 TaskGroup (https://docs.python.org/3/library/asyncio-task.html#asyncio.TaskGroup) 是一种结构化并发 (https://en.wikipedia.org/wiki/Structured_concurrency) 形式，它让开发者能够以简洁的方式创建多个并发任务。 ```python async with asyncio.TaskGroup() as tg: tg.create_task(run()) tg.create_task(run()) # 等待所有任务完成 ``` 假设我们想在后台等待某种信号来中断任务组的执行，这在 asyncio 中看似简单，但实际上实现起来有点笨拙： ```python class Interrupt(Exception): ... with suppress(Interrupt): async with asyncio.TaskGroup() as tg: tg.create_task(run()) tg.create_task(run()) if await wait_for_signal(): raise Interrupt() ``` 这段代码可行是因为在任务组内引发异常会导致其他任务被取消。自定义的 `Interrupt` 异常作为 `ExceptionGroup` 的一部分被引发，然后由 `contextlib.suppress` (https://docs.python.org/3/library/contextlib.html#contextlib.suppress) 过滤，从而实现优雅退出。 > `suppress` 对 `ExceptionGroup` 的处理方式是 3.12 版本中另一个被忽视的特性。我是在研究本文时偶然发现这个变化的。 新的 `TaskGroup.cancel` (https://docs.python.org/3.15/library/asyncio-task.html#asyncio.TaskGroup.cancel) 方法让这个过程变得简单得多： ```python async with asyncio.TaskGroup() as tg: tg.create_task(run()) tg.create_task(run()) if await wait_for_signal(): tg.cancel() ``` 与之前不同，它如此简单，几乎无需解释。它只是取消整个组而不引发任何异常。 ## 上下文管理器改进 装饰器的编写出奇地困难，以至于它已成为流行的面试题。但你知道上下文管理器也可以用作装饰器吗？ ```python @contextmanager def duration(message: str) -> Iterator[None]: start = time.perf_counter() try: yield finally: print(f"{message} elapsed {time.perf_counter() - start:.2f} seconds") ``` 这里我有一个非常常用的上下文管理器，用于打印代码块所花费的时间。自 Python 3.3 以来，我们也可以直接将其用作装饰器： ```python @duration('workload') def workload(): ... # 或者简单地作为包装器 duration('stuff')(other_workload)(...) ``` 虽然很方便，但在某些情况下它完全无法工作： ```python @duration('async workload') async def async_workload(): ... @duration('generator workload') def workload(): while True: yield ... ``` 迭代器、异步函数和异步迭代器在这里效果不佳，因为它们与标准函数的语义不同。调用它们时会分别立即返回生成器对象、协程函数和异步生成器对象。因此装饰器立即完成，而不是覆盖被包装对象的整个生命周期。 这是我多次遇到的不幸问题，对于普通装饰器来说也经常是个问题。但在 3.15 中情况发生了变化，现在 `ContextDecorator` 会检查它所包装函数的类型，并确保装饰器覆盖整个生命周期。 在我看来，这使得上下文管理器成为创建装饰器的最佳方式！它避免了一些常见的陷阱，并提供了更简洁的语法。我建议更多人开始以这种方式使用它。 ## 线程安全迭代器 迭代器是现代 Python 的基础之一。迭代器类型允许我们将数据源与数据消费者分离，如下所示，从而得到更清晰的抽象： ```python lazy from typing import Iterator def stream_events(...) -> Iterator[str]: while True: yield blocking_get_event(...) events = stream_events(...) for event in events: consume(event) ``` 但当使用线程或自由线程时，这种抽象就失效了。迭代器默认不是线程安全的，因此我们可能会看到值被跳过或内部迭代器状态损坏。 在 3.15 中，这通过 `threading.serialize_iterator` (https://docs.python.org/3.15/library/threading.html#threading.serialize_iterator) 得到解决。我们只需将原始迭代器用此函数包装即可： ```python import threading events = threading.serialize_iterator(stream_events(...)) with ThreadPoolExecutor() as executor: fut1 = executor.submit(consume, events) fut2 = executor.submit(consume, events) ``` 还有 `threading.synchronized_iterator` (https://docs.python.org/3.15/library/threading.html#threading.synchronized_iterator) 装饰器，它只是将 `threading.serialize_iterator` 应用于生成器函数的结果。 最后，我们还有 `threading.concurrent_tee` (https://docs.python.org/3.15/library/threading.html#threading.concurrent_tee)，它不会分割值，而是将值复制到多个迭代器中： ```python source1, source2 = threading.concurrent_tee(squares(10), n=2) with ThreadPoolExecutor() as executor: fut1 = executor.submit(consume, source1) fut2 = executor.submit(consume, source2) ``` 在这些工具出现之前，我们主要依赖 `Queue` (https://docs.python.org/3/library/queue.html) 来同步线程间的消费。有了这些新功能，我们可以在多线程代码中避免改变抽象层。 ## 额外特性 去年我只重点介绍了 3 个特性，但今年有更多更新让我感兴趣。以下是另外两个也许影响较小但仍然非常有趣的变化。 ### Counter 的异或操作 `collections.Counter` (https://docs.python.org/3/library/collections.html#collections.Counter) 是一个非常实用的类。它可以轻松统计离散事件的频率。它的行为与 `dict[KeyType, int]` 非常相似，但带有大量有用的操作： ```python c = Counter(a=3, b=1) d = Counter(a=1, b=2) print(f"{c + d = }") # 两个计数器相加: c[x] + d[x] print(f"{c - d = }") # 相减（仅保留正数） ``` 输出： ``` Counter(a=4, b=3) Counter(a=1, b=0) ``` 但它也有一些更奇怪的操作： ```python print(f"{c & d = }") # 交集: min(c[x], d[x]) print(f"{c | d = }") # 并集: max(c[x], d[x]) ``` 输出： ``` Counter(a=1, b=1) Counter(a=3, b=2) ``` 可以将 `Counter` 理解为也代表一组离散的对象。因此在我们例子中，本质上是在做： ``` {a_0, a_1, a_2, b_0} & {a_0, b_0, b_1} == {a_0, b_0} {a_0, a_1, a_2, b_0} | {a_0, b_0, b_1} == {a_0, a_1, a_2, b_0, b_1} ``` 在 3.15 中，我们还可以将异或操作加入列表： ```python c = Counter(a=3, b=1) d = Counter(a=1, b=2) c ^ d == c | d - c & d == Counter(a=3, b=2) - Counter(a=1, b=1) == Counter(a=2, b=1) ``` 同样，用之前的符号表示最佳： ``` {a_0, a_1, a_2, b_0} ^ {a_0, b_0, b_1} == {a_1, a_2, b_1} ``` 我将其放在额外部分是因为我从未在 `Counter` 上使用过集合操作，而且我很难为异或操作想到一个具体的用例。但我很欣赏开发者为了完整性而添加它。 ### 不可变 JSON 对象 随着 3.15 中 `frozendict` (https://peps.python.org/pep-0814/) 的加入，我们现在能够以不可变（可哈希）的形式表示所有 JSON 类型（数组、布尔值、浮点数、null、字符串、对象）。 对 `json.load` (https://docs.python.org/3.15/library/json.html#json.load) 和 `json.loads` (https://docs.python.org/3.15/library/json.html#json.loads) 进行了修改，增加了 `array_hook` 参数，与现有的 `object_hook` 参数互补。这让我们可以直接将 JSON 对象解析成这种形式： ```python json.loads('{"a": [1, 2, 3, 4]}', array_hook=tuple, object_hook=frozendict) == frozendict({'a': (1, 2, 3, 4)}) ``` 由 Pelican (https://getpelican.com/) 强力驱动，它充分利用了 Python (https://www.python.org/)。主题来自 Smashing Magazine (https://www.smashingmagazine.com/2009/08/designing-a-html-5-layout-from-scratch/)，感谢！