Pythonアプリケーションでの異常系表現を形式化したい

こんにちは。Data Management Group（以下DMG）の宮崎です。主にデータパイプライン開発やWebサーバサイド開発に携わりつつ、CIやデータ基盤の改善にも取り組んでいます。

DMGでは機械学習を利用するサーバーアプリケーションやデータパイプラインの実装でPythonを利用しています。本稿は、これらを高速・高品質に実装する上で悩まされた、Pythonアプリケーションコードにおける異常系の取り扱いの話になります。

モチベーション

Pythonにおける異常系表現の基本は例外ですが、アプリケーションコードにおける例外の取り扱いはコード設計全体に影響を与える一方、納得できる形がなかなか見つからず苦悩してきました。また、近年ではCoding Agentが無闇に広いtryスコープを取り、そのスコープ内から例外を投げるコードを書くという悩みも追加され（lint ruleで軽減可能ではある）、例外にとらわれず異常系表現を形式化したい気持ちが強まっていました。

本稿の前提

Python 3.12+
pyright 1.1.407+
- reportMatchNotExhaustiveが有効

本稿の範囲

異常系の表現について取り扱っていますが、複数のエラーを合成する所謂Applicativeな表現/実装方法は扱っていません。

Pythonの例外システムへの課題感

ある関数がどの例外を送出し得るかやcatchされていない例外がないかが静的に検証されないことから、これらの情報を得るためには呼び出すコードの中身を把握する必要があります。その上、コードを全て読んだとしてもその全てを常に正しく認知できるわけではありません。

個人的には、例外は型システムの外側で特殊な制御フローを実現するための機構だと捉えています。そして、型システムという縛りがないことと例外でしかできない制御フローがあるがゆえに、適切なガイドラインを策定しないと本来不要な複雑性をプロジェクトに持ち込んでしまいます。

関数シグネチャだけでは判断できないことがたくさんある

async def create_user(args: UserFactoryArgs) -> User:
    # どんな例外が送出され得る？どんなテストケースを作ればよい？
    # 送出する例外を追加・削除・変更した際、呼び出し元が壊れていないことをどうやって保障する？
    # 呼び出し側がloggingする必要がある？tracebackは書き出されているか？

どのようにアプローチするか

必ずしもコード上の工夫で対応しなければならないわけではありませんが、今回はコード上での工夫として以下の3つから検討します。

Eitherのようなコンテナを作る
共用体で状態列挙
契約プログラミング的例外制御

なお、これらのアプローチは互いに排他的ではなくレイヤや用途によって組み合わせて採用することもできます。本稿では比較のために 3 つに分けて整理します。

アプローチ1: Eitherのようなコンテナを作る

ScalaのEitherやRustのResultを模倣したclassを用意し、標準ライブラリやサードパーティライブラリ呼び出し箇所で例外をmappingするアプローチです。

実装方法は複数ありますが一例はこのような形です。

@dataclass
class Left[L]:
    value: L

    def get_or_else[R](self, default: R) -> R:
        return default

    # 様々なmethod
    ...


@dataclass
class Right[R]:
    value: R

    def get_or_else(self, default: R) -> R:
        return self.value

    # 様々なmethod
    ...

type Either[L, R] = Left[L] | Right[R]

使えなくはないのですが、Higher-Kinded Typeを扱えないPythonでは抽象度の高い関数の実装はすぐ限界に突き当たりますし、Haskellのdo notationやScalaのfor expressionsのような便利構文がないため、Railway Oriented Programming のように「成功パスだけを書いていくと途中の失敗が自動的に伝播する」スタイルも取りにくいです。一応decoratorとgeneratorを組み合わせることでそれらしい見た目にはできますが、失敗する可能性のある関数を全てgeneratorに変えるというのも現実的ではなさそうです。

また、2つの状態は表現できますが、発生する可能性のある例外を全てEither.Leftに入れるのか？という問題が残り、Either単体では解決しきれなさそうです。

Pros

正常系と異常系の存在が戻り値の型で示される
汎用的な様々なmethodを使用できる

Cons

Pythonの型システムでは抽象度の高いmethod実装の限界が浅い
do notationやfor expressionsのような糖衣構文がなく記述は冗長になりやすい
3つ以上の状態を表現するためには別の手段と組み合わせる必要がある

アプローチ2: 共用体で状態列挙

とり得る状態ごとに型を作って標準ライブラリやサードパーティライブラリ呼び出し箇所で例外をmappingするアプローチです。不正な状態を許さない型付けを行おうとすると、名義的直和型を持たないPythonでは自ずと共用体を使用することになり、例外の文脈でなくとも使用するテクニックです。

匿名classのないPythonではあまりないとは思いますが、複数の状態で値が同じ構造をとる場合は判別可能共用体にすれば良いです。また、細かい点ではありますが、単純な共用体では共用体を受け取る側が、パターンマッチのために共用体に含まれる全てのclassをimportする必要があります。ここをシンプルにするために判別可能共用体にしても良いと思います。

予約処理をテーマにした判別可能共用体のイメージです（抜粋なのでそのままでは動きません）。

class ReservationResultTag(IntEnum):
    """予約の結果のタグ"""

    # 正常
    RESERVED = auto()
    """予約確定"""
    WAITLISTED = auto()
    """キャンセル待ち登録"""

    # ビジネスドメイン解釈系
    SLOT_UNAVAILABLE = auto()
    """予約枠なし"""
    DUPLICATE_RESERVATION = auto()
    """重複予約"""
    PAST_DEADLINE = auto()
    """予約期限超過"""

    # テクニカルエラー
    TECHNICAL_FAILURE = auto()

type ReservationResult = (
    tuple[Literal[ReservationResultTag.RESERVED], Reserved]
    | tuple[Literal[ReservationResultTag.WAITLISTED], Waitlisted]
    | tuple[Literal[ReservationResultTag.SLOT_UNAVAILABLE], SlotUnavailable]
    | tuple[Literal[ReservationResultTag.DUPLICATE_RESERVATION], DuplicateReservation]
    | tuple[Literal[ReservationResultTag.PAST_DEADLINE], PastDeadline]
    | tuple[Literal[ReservationResultTag.TECHNICAL_FAILURE], TechnicalFailure]
)

def handle_reservation_result(result: ReservationResult) -> str:
    """予約結果をハンドリングする例"""
    match result:
        case (ReservationResultTag.RESERVED, reserved):
            ...
        case (ReservationResultTag.WAITLISTED, waitlisted):
            ...
        case (ReservationResultTag.SLOT_UNAVAILABLE, unavailable):
            ...
        case (ReservationResultTag.DUPLICATE_RESERVATION, duplicate):
            ...
        case (ReservationResultTag.PAST_DEADLINE, past):
            ...
        case (ReservationResultTag.TECHNICAL_FAILURE, failure):
            ...

上記はクリーンアーキテクチャなどで見られるレイヤ構造を前提としており、TECHNICAL_FAILUREにはDB接続エラーや外部APIタイムアウトといった、ビジネスルールとは無関係なインフラ起因の失敗をまとめて載せる運用を想定しています。infra層では生の例外をログ・監視（例: Sentryへの送出）に渡しつつ、domain層にはTechnicalFailureとして「テクニカルエラーが発生した」という事実だけを返す構造にすると、domain側がビジネスルールのみに集中できます。

また、pyrightでreportMatchNotExhaustiveを有効化しておくと新しいタグを追加した際に、対応するcaseを追加し忘れたmatchに対して警告を出してくれます。これにより、状態増加に伴うハンドリングの更新漏れを静的解析で広く検出できます。

Pros

共用体の使用はPythonにおける「普通の記述」の範疇であり取り入れやすい
n種類の状態を表現する手法として汎用的
async関数の特別扱いが不要

Cons

例外による制御フローと比較し記述が長大になりやすい

アプローチ3: 契約プログラミング的例外制御

指定した例外以外は契約違反例外にmapするようなdecoratorをつけることで、送出される例外を「指定例外 + 契約違反例外」のみに強制するアプローチです。

@raises(
    SlotUnavailableError,
    DuplicateReservationError,
    PastDeadlineError,
    MembershipRequiredError,
)
def make_reservation(...) -> ...:
    ...

静的解析は諦めているものの、関数の中身までは把握せずともハンドリングする必要がある例外がわかり、Python標準である例外システムをそのまま利用できるのが利点です。

静的解析を諦める以外でのConsは適切にdecoratorをつけるのが難しいことで、decorator関数の中身はおおよそ次のようになり、いたるところにつけるのはオーバーヘッドが気になります。現実的な対応策としてinfra層の関数に限り適用するなどが考えられます。

# 非同期関数は異なる実装が必要
def raises(*allowed_exceptions: type[Exception]) -> Callable[[Callable[P, R]], Callable[P, R]]:
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            try:
                return func(*args, **kwargs)
            except allowed_exceptions:
                raise
            except Exception as e:
                raise ContractViolationError(
                    original_exception=e,
                    function_name=func.__name__,
                ) from e

        return wrapper

    return decorator

Pros

例外制御フローを扱える

Cons

静的検証の欠如
送出され得る例外を真に強制するには全ての関数にdecoratorをつける必要がある

どのアプローチを採用するか

私の関わるプロジェクトではアプローチ2（共用体もしくは判別可能共用体による実装）を採用しています。以下の観点から採用する価値があると判断しました。

静的解析 → pyrightによりハンドリングの網羅性が保証
テスト戦略 → 機械的に状態空間をカバーしやすい（共用体を成す個別の型に対してテストケースを用意）
レビュー容易性 → 型から型への状態遷移に着目すればよく、見えない要素に思考を馳せる必要性が低い
Coding Agent親和性 → 要ハンドリングなケースが必然的にコンテキストに加わる
新たな悩みの種にならないか → 例外の文脈でなくとも使用する機能でありリスク低

Eitherは糖衣構文や非同期関数の抽象化がないこと、契約プログラミング的例外制御は「どの範囲の関数にデコレータを付けるか」という新たな設計判断が必要になることから、それぞれ別種の悩みを生みがちです。その点、共用体はそれ自体がPythonにおける「普通の型付け」であり、導入にあたって新しいメンタルモデルを強要しないのが大きな利点でした。Coding Agentが上手く扱えているというのも重要な点で、例外の世界と比較してコンテキスト的にも有利なうえ静的解析が付いてくるのでコードの自動生成が捗ります。