🐍

Pythonの型ヒントと共に進化するコード(#11: コレクション抽象型 Mapping)

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


前回の 10 日目では dataclasses を導入してデータと振る舞いをカプセル化しました。

Address クラスは不変のオブジェクトになり、安全性は上がりました。

ただ、前回の最後で触れたように、API レスポンスから Address を生成する部分にはまだ課題が残っています。

今回の課題:API レスポンスとの直接結合

main.py で API レスポンスから Address インスタンスを生成している部分を見てみます。

main.py の一部
address = Address(**response.json())

response.json() の戻り値(辞書)を直接 Address のコンストラクタに展開しています。

シンプルで良さそうに見えますが、実はこの書き方には罠があります。

1. 余分なフィールドでエラーになる

API のレスポンスに Address クラスが定義していないフィールドが含まれていたらどうなるか。たとえば latitudelongitude が追加されたとします。

payload = {
    "zipcode": "1000001",
    "prefecture": "東京都",
    # ... 必要なフィールド ...
    "latitude": 35.6762,   # Address には存在しないフィールド
    "longitude": 139.6503,
}

Address(**payload)  # TypeError: 予期しないキーワード引数 'latitude'

dataclass のコンストラクタは定義されていないキーワード引数を受け取ると TypeError を発生させます。外部 API にフィールドが追加されただけでこちらのコードが壊れてしまう。正直これはかなり困ります。

2. 外部スキーマの変更がドメインモデルに直接影響する

もし将来的に、

  • API 側で prefecturepref に変更されたら?
  • zipcode が整数で返されるようになったら?

これらの変更がアプリケーション内の Address を使っているすべての箇所に波及します。

外部 API のスキーマ(ゆるい世界)とアプリケーション内部のドメインモデル(厳密な世界)は別物です。その間には変換を担う境界が必要です。

処方箋:変換メソッドと Mapping で境界を作る

この問題を解決するため Address クラスに unmarshal_payload という変換メソッドを追加します。このメソッドが外部データから Address への変換を一手に引き受けます。

from typing import Any, Mapping

@dataclass(frozen=True, slots=True)
class Address:
    # ... フィールド定義 ...

    @classmethod
    def unmarshal_payload(cls, payload: Mapping[str, Any]) -> "Address":
        """APIレスポンス内の必要なキーだけを抽出してインスタンス化する"""
        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"]),
        )

図にするとこんなイメージです。

変換メソッドが境界となって、外部スキーマの変更からドメインモデルを守ります。

さて、この変換メソッドの引数に注目してください。

dict[str, Any] ではなく Mapping[str, Any] を使っています。

なぜ dict ではなく Mapping なのか

Mappingcollections.abc で定義された抽象型で、キーで値を読み取るという振る舞いだけを持ちます。
dict と違って .pop()["key"] = value といった書き換え操作は定義されていません。

この性質を利用して、引数の型に Mapping[str, Any] を指定すればこの関数は渡されたデータを書き換えないという契約を型で表現できます。

具体的には次のとおりです。

1. dict だと書き換えが素通りする

dict[str, Any] を引数にすると、関数の中でこういうコードを書いても型チェックは素通りします。

def bad_func(payload: dict[str, Any]) -> None:
    # 読み取りだけのつもりだったのに…
    payload["zipcode"] = "0000000"  # 型的には合法

これだと、読み取るだけの関数のはずが、いつの間にか副作用を持ち始めていたという事故を防げません。

2. Mapping なら型チェッカーが止めてくれる

一方、引数の型を Mapping[str, Any] にするとどうなるでしょうか。

from collections.abc import Mapping

def good_func(payload: Mapping[str, Any]) -> None:
    _ = payload["zipcode"]  # 読み取りは OK
    payload["zipcode"] = "0000000"  # ← mypy がエラーにしてくれる

mypy でチェックすると書き換え行はこんな感じで怒られます。

error: Unsupported target for indexed assignment ("Mapping[str, Any]")

ポイントはここです。

  • 呼び出し側は相変わらず dict を渡してよい
  • しかし受け取る側の関数は「自分は読み取り専用として扱う」と型で宣言できる
  • その宣言に反したコード(payload["k"] = ....pop() など)を書いた瞬間、型チェッカーが止めてくれる

つまり Mapping は実体がイミュータブルだから変更できないのではなく、「この関数が変更しないという約束を型で自分に課している」という役割を持つ抽象型です。

unmarshal_payload の目的は外部データを読み取って Address を生成することなので、この契約がぴったり当てはまります。

3. 柔軟性の向上

Mapping は抽象型なので、dict に限らず「キーで値を読む」という振る舞いを満たすものなら何でも受け取れます。

たとえば将来、dict 以外の dict-like オブジェクト(ChainMapMappingProxyType など)で渡したくなっても、引数が Mapping なら受け入れられます。

具体的な実装ではなく「キーで値を読める」という振る舞いに依存する。これは 8 日目で紹介した Protocol と同じ考え方ですね。

MappingProxyType とは

MappingProxyTypetypes モジュールで提供される読み取り専用の dict ビューです。元の dict をラップして読み取りはできるが書き換えはできないオブジェクトを作れます。

from types import MappingProxyType

config = {"host": "localhost", "port": "8080"}
frozen = MappingProxyType(config)

print(frozen["host"])  # OK
frozen["host"] = "example.com"  # TypeError: 'mappingproxy' object does not support item assignment

Mapping が型レベルで書き換えない契約なのに対し、MappingProxyTypeランタイムで本当に書き換えられない実装です。レイヤーが違うものとして理解すると整理しやすくなります。

変換メソッド(unmarshal_payload) 導入のメリット

続いて、前述の変換メソッド(unmarshal_payload)を使うと main.py は以下のように変わります。

# 変更前
address = Address(**response.json())

# 変更後
address = Address.unmarshal_payload(response.json())

得られるメリットは大きいです。

  • 余計なフィールドを無視できる

    • payloadlatitudelongitude があっても問題ありません。unmarshal_payload で使うキーだけ取り出すので TypeError は起きません。
  • 型変換やバリデーションを書く場所ができる

    • API が zipcode を整数で返すようになっても str(payload["zipcode"]) にしておけば呼び出し側は変更不要です。
  • 外部スキーマの変更からドメインモデルを守れる

    • API 側で prefecturepref に変わっても unmarshal_payload の中を書き換えるだけで済みます。Address を使っている他のコードには一切手を入れなくて済む。

unmarshal_payloadメソッド は「外部のゆるい世界」と「内部の厳密な世界」の境界として機能します。この境界があるかないかで保守性がまるで違います。

コレクション抽象型ファミリー

ここまで Mapping を使ってきましたが、collections.abc には同様の抽象型が他にもいくつか用意されています。せっかくなのでここで主要なものを整理しておきます。

主要なコレクション抽象型

説明 代表的な具象型
Mapping[K, V] キー → 値のマッピング(読み取り前提) dictTypedDict も実体は dict)
MutableMapping[K, V] キー → 値の変更可能マッピング dict
Sequence[T] インデックスアクセス可能(読み取り前提) list, tuple, str
MutableSequence[T] インデックスアクセス可能な変更可能コレクション list
Iterable[T] for ループで回せるもの全般 ほぼすべてのコレクション
Collection[T] for で回せて len() が使えて in で含まれるか調べられる list, set, dict

使用例:SequenceIterable

SequenceIterable の使い分けも見ておきましょう。複数の住所を処理する関数を考えます。

from collections.abc import Sequence, Iterable

def format_addresses(addresses: Sequence[Address]) -> list[str]:
    """複数の Address をフォーマットして返す"""
    return [addr.full_address() for addr in addresses]

def count_by_prefecture(addresses: Iterable[Address]) -> dict[str, int]:
    """都道府県ごとの住所数をカウントする"""
    counts: dict[str, int] = {}
    for addr in addresses:
        counts[addr.prefecture] = counts.get(addr.prefecture, 0) + 1
    return counts

Sequence を使う場面

Sequence は順序があり、インデックスでアクセスできることを表します。list でも tuple でも受け取れます。

# list でも tuple でも OK
format_addresses([addr1, addr2, addr3])
format_addresses((addr1, addr2, addr3))

Iterable を使う場面

Iterablefor で回せることだけを要求します。listtupleset、ジェネレータなど何でも受け取れます。一番緩い条件ですね。

# ジェネレータも OK
def address_generator() -> Iterable[Address]:
    yield addr1
    yield addr2

count_by_prefecture(address_generator())

collections.abc には他にもコレクション抽象型が用意されています。興味があれば公式ドキュメントを覗いてみてください。

コードの進化:スクリプトへの適用

それでは unmarshal_payloadMapping をスクリプトに適用しましょう。

models.py (Address クラス周りの抜粋)

from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, ClassVar

@dataclass(frozen=True, slots=True)
class Address:
    API_PATH: ClassVar[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"]),
        )

main.py (変更後)

# main.py
def fetch_and_format_address(
    zipcode: ZipCode,
    include_kana: bool,
    http_client: HttpClient,
    headers: Headers | None = None,
) -> str | None:
    api_url = f"https://api.zipcode-jp.example{Address.API_PATH}"

    try:
        response = http_client.post(api_url, json={"zipcode": zipcode}, headers=headers)
        # ...

        # 👉 unmarshal_payload経由でインスタンス化
        address = Address.unmarshal_payload(response.json())

        full_address = address.full_address()
        # ...

得られたものと、次の課題

今回は変換メソッドと Mapping によって外部 API とドメインモデルの間に明確な境界を設けました。API スキーマの変更でモデルが壊れる心配から解放されます。

ただコードを見渡すとまだ気になる箇所があります。

# この "https://api.zipcode-jp.example" は変更されるべきでない定数では?
api_url = f"https://api.zipcode-jp.example{Address.API_PATH}"

# ステータスコード 200 も定数として名前を付けたい
if response.status_code != 200:
    ...

これらの値は絶対に変更されるべきでない定数です。現状では単なる変数やリテラルとして扱われており、誤って書き換えてしまうリスクがあります。

次回予告

次回は Final を導入して定数を型システムで保護します。

処方後のコードはこちら

models.py

# models.py
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, ClassVar, NewType, TypedDict

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

@dataclass(frozen=True, slots=True)
class Address:
    API_PATH: ClassVar[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"]),
        )

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

class FormattedAddressWithKanaDict(FormattedAddressDict):
    full_address_kana: str

http_client.py

# http_client.py
import requests as requests_lib
from typing import Protocol

from models import Headers

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()

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

main.py

# main.py
import json
from collections.abc import Mapping
from typing import Any, cast

from models import ZipCode, Headers, Address, FormattedAddressDict, FormattedAddressWithKanaDict
from http_client import HttpClient, RequestsHttpClient


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

    api_url = f"https://api.zipcode-jp.example{Address.API_PATH}"

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

        # 👉 unmarshal_payload経由でインスタンス化
        payload = cast(Mapping[str, Any], response.json())
        address = Address.unmarshal_payload(payload)

        full_address = address.full_address()

        result: FormattedAddressDict = {
            "zipcode": zipcode,
            "full_address": full_address,
            "prefecture": address.prefecture,
            "city": address.city,
            "town": address.town,
        }

        if include_kana:
            result_with_kana: FormattedAddressWithKanaDict = {
                **result,
                "full_address_kana": address.full_address_kana(),
            }
            return json.dumps(result_with_kana, indent=2, ensure_ascii=False)

        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)

Discussion