Python 3.15：表舞台には立たなかった機能

もう一度この時期が訪れました。Python の新バージョンも間近です。Python 3.15.0b1 では機能の凍結（feature freeze）が行われ、今年後半に Python にどのような新功能が加わるかについては明確になっています。特に注目すべき大きな変更には、「非同期 import」（lazy imports）や私が以前にも取り上げた「タキオンプロファイラ（Tachyon profiler）」などがあります。

昨年は、Python 3.14 の较小的機能についても詳しく調査しましたところ、それらも PEP に掲げられる大規模な仕様変更と同じくらい興味深く、さらに多くの人々の関心を集めるに値することがわかりました。今年の状況も同じです。

asyncio.TaskGroup

の優雅なキャンセル機能

今回のリリースでは asyncio に関する変更点はそれほど多くはありません。主な新機能は、

TaskGroup

を「優雅に（gracefully）」安全にキャンセルできる点にあります。

TaskGroup

は構造化併行処理（structured concurrency）の一つの形態で、開発者が複数のタスクをクリーンな方法で同時に実行できるようにします。

async with asyncio.TaskGroup() as tg:
    tg.create_task(run())
    tg.create_task(run())
    # すべてのタスクが完了するのを待ちます

さて、ここではタスクグループの実行を中断するようなシグナルの受け取りをバックグラウンドで待機したいとしましょう。これは asyncio において単純なように思えますが、実際にはやや難解です。

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

でフィルタリングされて、結果的にプログラムが正常に終了します。

ここで使われている

suppress

と

ExceptionGroup

の動作は、3.12 版から見過ごされていた興味深い機能の一つです。この記事の調査中に偶然この事実を知りました。

新しい

TaskGroup.cancel

メソッドを導入するだけで、この処理は大幅に簡素化されます：

async with asyncio.TaskGroup() as tg:
    tg.create_task(run())
    tg.create_task(run())
    if await wait_for_signal():
        tg.cancel()

以前の手法とは異なり、これほど単純になったことでわざわざ説明する必要もほとんどありません。例外を発生させることなく、グループ全体をただキャンセルするだけです。

コンテキストマネージャ機能の改善

デコレータの作成は意外と難しく、「面接における定番の質問」とさえなっています。しかし、コンテキストマネージャも実はデコレータとして同時利用可能であることを知っていますか？

@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 よりも前から、これをそのままデコレータとして直接使用することも可能です。

@duration('workload')
def workload():
    ...

# もしくは単なるラッパー関数として
duration('stuff')(other_workload)(...)

確かに便利な仕組みですが、すべてのケースで動作するわけではありません：

@duration('async workload')
async def async_workload():
    ...

@duration('generator workload')
def workload():
    while True:
        yield ...

イテレータ、非同期関数、および非同期イテレータは、通常の関数とは異なるセマンティクスを持つため、これらのケースではうまく機能しません。これらを呼び出すと、直ちにそれぞれ「ジェネレーターオブジェクト」「コルーチン関数」「アシンクロジェネレーターオブジェクト」を返します。そのため、デコレータの処理が対象となる全体のスコープ（ライフサイクル）ではなく、呼び出し直後だけで終了してしまいます。

この問題は私自身も何回か遭遇した不運な事例ですが、通常のデコレータにおいてもよく見られる問題です。しかし Python 3.15 では状況が変わりました。

ContextDecorator

は包装される関数のタイプを確認し、デコレーション対象の整个なライフサイクルをカバーするように振る舞います。

私の見解では、これによりコンテキストマネージャが「デコレータを作るための最適な方法」として再評価されます。一般的な落とし穴を避けつつ、より洗練された構文を提供するためです。このアプローチを多くの人々が採用することを強く推奨します。

スレッド安全なイテレータ

イテレータは現代の Python の基盤の一つであり、イテレータ型により「データソース」と「データ消費者」を明確に分離できます。これにより、より洗練された抽象化が可能になります。

from typing import Iterator

def stream_events(...) -> Iterator[str]:
    while True:
        yield blocking_get_event(...)

events = stream_events(...)

for event in events:
    consume(event)

しかし、この抽象化はスレッド処理や「フリースレッド（free-threading）」環境下では破綻します。デフォルトのイテレータはスレッド安全ではないため、データがスキップしたり、内部状態が壊れたりする可能性があります。

Python 3.15 では

threading.serialize_iterator

を用いてこの問題が解決します。オリジナルのイテレータをこの関数でラップするだけで、問題が解消されます：

import threading

events = threading.serialize_iterator(stream_events(...))

with ThreadPoolExecutor() as executor:
    fut1 = executor.submit(consume, events)
    fut2 = executor.submit(consume, events)

また、ジェネレーター関数の結果に対して

threading.serialize_iterator

を自動的に適用するデコレータである

threading.synchronized_iterator

も用意されています。

さらに、複数のイテレータ間で値を複製（分岐させるのではなく）するための

threading.concurrent_tee

もあります：

source1, source2 = threading.concurrent_tee(squares(10), n=2)

with ThreadPoolExecutor() as executor:
    fut1 = executor.submit(consume, source1)
    fut2 = executor.submit(consume, source2)

これらのユーティリティが存在する以前には、マルチスレッド間での消費処理の同期は主にキュー（Queue）に頼っていました。これら新增された機能によって、マルチスレッド用コードであっても、既存の抽象化を変更する必要がなくなります。

余談：さらに興味深い機能

昨年は 3 つの機能を重点的に紹介しましたが、今年はこれよりも多くの新機能が発表されており、特に魅力的です。今回はインパクトはさほど大きくありませんが、それでも非常に興味深い 2 つの変更点をご紹介します。

Counter

クラスへの XOR 演算子の追加

collections.Counter

は非常に有用なクラスです。離散な発生頻度を簡潔にカウントできます。これは文字列マップ（

dict[KeyType, int]

）のような振る舞いをしますが、さらに多くの有用な演算子を内蔵しています。

c = Counter(a=3, b=1)
d = Counter(a=1, b=2)
print(f"{c + d = }")  # 2 つの Counter を加算：各要素について c[x] + d[x]
print(f"{c - d = }")  # 減算（正値のみ保持）

出力例：

Counter(a=4, b=3)
Counter(a=1, b=0)

さらに奇妙ですが面白い演算もあります：

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

も離散なオブジェクトの集合（multiset）を表現できる」という考え方です。つまり、上記の例では本質的には以下のようになります：

{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}

Python 3.15 ではさらに「排他的和（XOR）」演算も追加されます：

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}

私はこれを「余談（bonus）」のセクションに載せました。なぜなら、私はこれまで Counter に対する集合演算を使用する機会はほとんどなかったし、「XOR」に特化した具体的な使用例を想像することが極めて困難だったからです。しかし、コンプリートネスのために開発者がこの機能を追加してくれたことに感謝します。

インミュータブルな JSON オブジェクト

Python 3.15 で

frozendict

が追加されることで、すべての JSON 型（配列、ブーリアン、浮動小数点数、null、文字列、オブジェクト）を「不変かつハシシャブル（hashable）」な形で表現できるようになりました。

さらに

json.load

および

json.loads

に、

object_hook

パラメータと対となる

array_hook

パラメータが追加されました。これにより、JSON オブジェクトを以下のような不変形に直接パースできます：

json.loads('{"a": [1, 2, 3, 4]}', array_hook=tuple, object_hook=frozendict) 
# == frozendict({'a': (1, 2, 3, 4)})