🐍

Pythonの型ヒントと共に進化するコード(#18: Type Narrowing)

に公開
これまでの連載記事


前回は型安全なデコレータを実装し、関数のシグネチャをそのまま保持できるようになりました。

ここまでで型システムの表現力はかなり上がっています。

今回は Union 型を扱うときに欠かせない Type Narrowing(型の絞り込み)を取り上げます。
5 日目UnionOptional を紹介しましたが、あのときは複数の型を許容できるという話でした。今回はその逆で複数の型のうち、どれなのかを確定させる話です。

今回の課題:API レスポンスが 2 種類ある

ある日、外部の住所検索 API に新しい仕様(破壊的変更)が追加されたことを知りました。
これまでは正常時に住所データを返すだけでしたが、エラー時には専用のエラーレスポンスを返すようになったようです。急いで修正が必要です。

判明した新しいAPIレスポンスの仕様
# 正常時
{"zipcode": "1000001", "prefecture": "東京都", ...}

# エラー時(郵便番号が見つからない場合など)
{"error_code": "NOT_FOUND", "message": "指定された郵便番号は存在しません"}

これを型で表現しようとすると、レスポンスは Address | ApiError という Union 型になりそうですね。

# 正常時のレスポンスを表す型(これまで使ってきたもの)
@dataclass(frozen=True, slots=True)
class Address:
    zipcode: str
    prefecture: str
    # ... 他のフィールド省略

    def full_address(self) -> str:
        return self.prefecture + self.city + self.town

# エラー時のレスポンスを表す型(今回追加)
@dataclass(frozen=True, slots=True)
class ApiError:
    error_code: str
    message: str

問題は、この Union 型をどうやって扱うかです。

API レスポンスを受け取って処理する関数を考えてみます。正常時は住所を文字列で返し、エラー時はエラーメッセージを返すとしましょう。

def handle_response(response: Address | ApiError) -> str:
    """APIレスポンスを処理して文字列を返す"""
    # 引数で受け取る response は Address かもしれないし ApiError かもしれない
    # ここで response.full_address() を呼ぶと…?
    return response.full_address()  # エラー: ApiError には full_address がない

型チェッカーは怒ります。当然です。ApiError には full_address メソッドが存在しないのですから。

isinstance で型を絞り込む

Python には isinstance という組み込み関数があります。オブジェクトが特定のクラスのインスタンスかどうかを実行時にチェックできます。

def handle_response(response: Address | ApiError) -> str:
    if isinstance(response, Address):
        # ここでは response は Address 型として扱える
        return response.full_address()
    else:
        # ここでは response は ApiError 型として扱える
        return f"エラー: {response.message}"

isinstance でチェックした後、型チェッカーは条件分岐の中で型を絞り込んでくれます。これが Type Narrowing(型の絞り込み)です。

if ブロックの中では responseAddress として扱えるし、else ブロックの中では ApiError として扱えます。実行時のチェックと型チェッカーの推論が連動しているわけです。

TypeGuard:独自の判定ロジックを型に伝える

isinstance で十分なケースが多いですが、判定ロジックを関数として切り出したいこともあります。たとえば「エラーレスポンスかどうか?」を判定する処理を複数箇所で使い回したい場合です。

素朴に書くとこうなります。

def is_api_error(response: Address | ApiError) -> bool:
    """エラーレスポンスかどうかを判定する"""
    return isinstance(response, ApiError)

この関数は実行時には正しく動きます。でも型チェッカーには何も伝わりません。

def handle_response(response: Address | ApiError) -> str:
    if is_api_error(response):
        # 型チェッカーは response がまだ Address | ApiError だと思っている
        return f"エラー: {response.message}"  # エラー: Address には message がない
    return response.full_address()

is_api_errorTrue を返したのだから responseApiError のはずです。でも型チェッカーはそれを知りません。戻り値の型が bool だと真偽値の情報しか伝わらないからです。

ここで登場するのが TypeGuard です。

from typing import TypeGuard

def is_api_error(response: Address | ApiError) -> TypeGuard[ApiError]:
    return isinstance(response, ApiError)

TypeGuard[ApiError] という戻り値の型はこの関数が True を返したら、引数は ApiError として扱ってよいという意味です。呼び出し側で使ってみましょう。

def handle_response(response: Address | ApiError) -> str:
    if is_api_error(response):
        # response は ApiError として扱える
        return f"エラー: {response.message}"
    # ⚠️ ここは「TypeGuard だけでは else 側が絞れない」ことを示すため、あえて型エラーになる例です
    return response.full_address()  # エラー: ApiError には full_address がない

実はこれだけだと else 側が絞り込まれていません。
TypeGuard は if ブロックの中だけを ApiError に絞り込みますが、else 側は Address | ApiError のままになっており絞り込みできていない状態なのです。

TypeIs:else 側も絞り込む

これを解決する手段があります。 Python 3.13 で追加された TypeIs です。
(それ以前は typing_extensions から使えます)

from typing import TypeIs  # Python 3.13+
# from typing_extensions import TypeIs  # Python 3.12以前

def is_api_error(response: Address | ApiError) -> TypeIs[ApiError]:
    return isinstance(response, ApiError)

TypeIs を使うと else 側も絞り込まれます。

def handle_response(response: Address | ApiError) -> str:
    if is_api_error(response):
        # response は ApiError
        return f"エラー: {response.message}"
    # response は Address(ApiError が除外された)
    return response.full_address()  # OK!

TypeIs条件を満たさなければその型ではないという否定側の情報も型チェッカーに伝えます。Union 型を分岐させるときはこちらのほうが便利です。

TypeGuard と TypeIs の使い分け

特徴 TypeGuard TypeIs
if 側の絞り込み
else 側の絞り込み ×
導入バージョン Python 3.10 Python 3.13

基本的に TypeIs を使っておけば問題ないと考えます。
(Python 3.13 以前のバージョンでも typing_extensions からインポートすれば使えます)

else 側も絞り込んでくれるので Union 型の網羅的な分岐処理が書きやすくなります。

TypeGuard は True のときだけ特定の型として扱いたい(False のときは絞り込み不要の)場合に使えば OK です。

コードの進化:Union 型を安全に扱う

では、連載で育ててきた住所検索コードに Type Narrowing を適用してみます。

API の仕様変更に対応するため、ApiError クラスと is_error_response 関数を追加してレスポンスを安全に分岐させます。

変更前

def fetch_and_format_address(
    zipcode: ZipCode,
    include_kana: bool,
    http_client: HttpClient,
    headers: Headers | None = None,
) -> str | None:
    api_url = f"{BASE_URL}{Address.API_PATH}"

    try:
        response = http_client.post(api_url, json={"zipcode": zipcode}, headers=headers)
        if response.status_code != HTTP_OK:
            print(f"Error: Failed to fetch address. Status: {response.status_code}")
            return None

        # レスポンスの中身が Address なのか ApiError なのか分からない
        payload = response.json()
        # ...

変更後

まず models.pyApiError を追加し、レスポンスを解析する関数を作ります。

models.py
from typing import TypeIs

@dataclass(frozen=True, slots=True)
class ApiError:
    error_code: str
    message: str

    @classmethod
    def unmarshal_payload(cls, payload: Mapping[str, Any]) -> ApiError:
        return cls(
            error_code=str(payload["error_code"]),
            message=str(payload["message"]),
        )

# レスポンスの型
type ApiResponse = Address | ApiError

def is_error_response(response: ApiResponse) -> TypeIs[ApiError]:
    """ApiError かどうかを判定する"""
    return isinstance(response, ApiError)

main.py では Union 型を安全に分岐させます。

main.py
def parse_response(payload: Mapping[str, Any]) -> ApiResponse:
    """APIレスポンスを解析して適切な型に変換する"""
    if "error_code" in payload:
        return ApiError.unmarshal_payload(payload)
    return Address.unmarshal_payload(payload)

def fetch_and_format_address(
    zipcode: ZipCode,
    include_kana: bool,
    http_client: HttpClient,
    headers: Headers | None = None,
) -> str | None:
    api_url = f"{BASE_URL}{Address.API_PATH}"

    try:
        response = http_client.post(api_url, json={"zipcode": zipcode}, headers=headers)
        if response.status_code != HTTP_OK:
            print(f"Error: Failed to fetch address. Status: {response.status_code}")
            return None

        payload = cast(Mapping[str, Any], response.json())
        api_response = parse_response(payload)

        # TypeIs による絞り込み
        if is_error_response(api_response):
            # api_response は ApiError
            print(f"API Error: {api_response.message}")
            return None

        # api_response は Address(ApiError が除外された)
        formatter = AddressFormatter()
        result = formatter.with_address(api_response).with_kana(include_kana).build()
        return json.dumps(result, indent=2, ensure_ascii=False)

    except Exception as e:
        print(f"An error occurred: {e}")
        return None

is_error_responseFalse を返した後、型チェッカーは api_responseAddress として認識します。full_address()AddressFormatter に渡しても型エラーになりません。

得られたもの

今回は Type Narrowing を使って Union 型を安全に扱う方法を紹介しました。

  • isinstance による基本的な絞り込み
  • TypeGuard で独自の判定ロジックを型に伝える
  • TypeIs で else 側も含めて絞り込む

Union 型は便利ですが、使う側で「どの型なのか」を確定させる必要があります。Type Narrowing はその確定作業を型チェッカーに伝える仕組みです。

次回予告

型ヒントを書いていると、こんな疑問が浮かんでくることはありませんか。

  • 意図した通りに型が絞り込まれているのか?
  • この時点では変数の型はどのように推論されているのか?
if is_error_response(api_response):
    # ここで api_response は本当に ApiError になってる?

こういうときに使えるのが reveal_type です。

reveal_type(api_response)  # Revealed type is "ApiError"

これは、型チェッカーが変数をどう認識しているのかを確認できるデバッグツールです。

次回はコラム回として reveal_type を紹介します。ナローイングがどこで効いているのか、Any がどこから混入したのか、Generics の型変数は何に束縛されたのか。こうした疑問を解決するデバッグ手法を紹介します。


処方後のコード

models.py

# models.py
from __future__ import annotations

from collections.abc import Mapping
from dataclasses import dataclass, replace
from typing import Any, ClassVar, Final, NewType, ReadOnly, Self, TypedDict, TypeIs # Python 3.13 前提

ZipCode = NewType("ZipCode", str)
type Headers = dict[str, str]

@dataclass(frozen=True, slots=True)
class Address:
    API_PATH: ClassVar[Final[str]] = "/v1/address"

    zipcode: str
    prefecture: str
    prefecture_kana: str
    city: str
    city_kana: str
    town: str
    town_kana: str

    def full_address(self) -> str:
        """都道府県・市区町村・町域を結合したフル住所を返す"""
        return self.prefecture + self.city + self.town

    def full_address_kana(self) -> str:
        """フル住所のカナ表記を返す"""
        return self.prefecture_kana + self.city_kana + self.town_kana

    @classmethod
    def unmarshal_payload(cls, payload: Mapping[str, Any]) -> Address:
        """APIレスポンスからAddressオブジェクトを生成する"""
        return cls(
            zipcode=str(payload["zipcode"]),
            prefecture=str(payload["prefecture"]),
            prefecture_kana=str(payload["prefecture_kana"]),
            city=str(payload["city"]),
            city_kana=str(payload["city_kana"]),
            town=str(payload["town"]),
            town_kana=str(payload["town_kana"]),
        )


@dataclass(frozen=True, slots=True)
class ApiError:
    error_code: str
    message: str

    @classmethod
    def unmarshal_payload(cls, payload: Mapping[str, Any]) -> ApiError:
        """APIレスポンスからApiErrorオブジェクトを生成する"""
        return cls(
            error_code=str(payload["error_code"]),
            message=str(payload["message"]),
        )


# APIレスポンスの型
type ApiResponse = Address | ApiError


def is_error_response(response: ApiResponse) -> TypeIs[ApiError]:
    """ApiError かどうかを判定する"""
    return isinstance(response, ApiError)


class FormattedAddressDict(TypedDict):
    zipcode: ReadOnly[str]
    full_address: ReadOnly[str]
    prefecture: ReadOnly[str]
    city: ReadOnly[str]
    town: ReadOnly[str]

class FormattedAddressWithKanaDict(FormattedAddressDict):
    full_address_kana: ReadOnly[str]

@dataclass(frozen=True, slots=True)
class AddressFormatter:
    _address: Address | None = None
    _include_kana: bool = False

    def with_address(self, address: Address) -> Self:
        return replace(self, _address=address)

    def with_kana(self, include: bool = True) -> Self:
        return replace(self, _include_kana=include)

    def build(self) -> FormattedAddressDict | FormattedAddressWithKanaDict:
        if self._address is None:
            raise ValueError("Address must be set before building.")

        base: FormattedAddressDict = {
            "zipcode": self._address.zipcode,
            "full_address": self._address.full_address(),
            "prefecture": self._address.prefecture,
            "city": self._address.city,
            "town": self._address.town,
        }

        if self._include_kana:
            with_kana: FormattedAddressWithKanaDict = {
                **base,
                "full_address_kana": self._address.full_address_kana(),
            }
            return with_kana

        return base

main.py

# main.py
from __future__ import annotations

import json
from typing import Any, Final, cast
from collections.abc import Mapping

from models import (
    ZipCode,
    Headers,
    Address,
    ApiError,
    ApiResponse,
    AddressFormatter,
    is_error_response,
)
from http_client import HttpClient, RequestsHttpClient

# 定数
BASE_URL: Final[str] = "https://api.zipcode-jp.example"
HTTP_OK: Final[int] = 200


def parse_response(payload: Mapping[str, Any]) -> ApiResponse:
    """APIレスポンスを解析して適切な型に変換する"""
    if "error_code" in payload:
        return ApiError.unmarshal_payload(payload)
    return Address.unmarshal_payload(payload)


def fetch_and_format_address(
    zipcode: ZipCode,
    include_kana: bool,
    http_client: HttpClient,
    headers: Headers | None = None,
) -> str | None:
    """郵便番号から住所を取得し、整形して返す"""

    api_url = f"{BASE_URL}{Address.API_PATH}"

    try:
        response = http_client.post(api_url, json={"zipcode": zipcode}, headers=headers)
        if response.status_code != HTTP_OK:
            print(f"Error: Failed to fetch address. Status: {response.status_code}")
            return None

        payload = cast(Mapping[str, Any], response.json())
        api_response = parse_response(payload)

        # TypeIs による絞り込み
        if is_error_response(api_response):
            # api_response は ApiError
            print(f"API Error: {api_response.message}")
            return None

        # api_response は Address(ApiError が除外された)
        formatter = AddressFormatter()
        result = formatter.with_address(api_response).with_kana(include_kana).build()

        return json.dumps(result, indent=2, ensure_ascii=False)

    except Exception as e:
        print(f"An error occurred: {e}")
        return None


# 実行例
if __name__ == "__main__":
    http_client = RequestsHttpClient()
    zipcode = ZipCode("1000001")
    result = fetch_and_format_address(
        zipcode, include_kana=True, http_client=http_client
    )
    if result is not None:
        print(result)

http_client.py

# http_client.py
from __future__ import annotations

import requests as requests_lib
from typing import Protocol

from models import Headers
from decorators import measure_time

type JsonObject = dict[str, object]


class HttpResponse(Protocol):
    @property
    def status_code(self) -> int: ...
    def json(self) -> object: ...


class HttpClient(Protocol):
    def post(self, url: str, json: JsonObject, headers: Headers | None = None) -> HttpResponse: ...


class RequestsResponse:
    def __init__(self, response: requests_lib.Response) -> None:
        self._response = response

    @property
    def status_code(self) -> int:
        return self._response.status_code

    def json(self) -> object:
        return self._response.json()


class RequestsHttpClient:
    def __init__(self) -> None:
        self._session = requests_lib.Session()

    @measure_time
    def post(self, url: str, json: JsonObject, headers: Headers | None = None) -> RequestsResponse:
        response = self._session.post(url, json=json, headers=headers)
        return RequestsResponse(response)

decorators.py

# decorators.py
from __future__ import annotations

import time
from functools import wraps
from typing import Callable


def measure_time[**P, R](func: Callable[P, R]) -> Callable[P, R]:
    """関数の実行時間を計測し、標準出力に表示するデコレータ"""
    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        start_time = time.perf_counter()
        result = func(*args, **kwargs)
        end_time = time.perf_counter()
        print(f"Finished '{func.__name__}' in {end_time - start_time:.4f} secs")
        return result
    return wrapper

Discussion