Pythonの型ヒントと共に進化するコード(#11: コレクション抽象型 Mapping)
これまでの連載記事
前回の 10 日目では dataclasses を導入してデータと振る舞いをカプセル化しました。
Address クラスは不変のオブジェクトになり、安全性は上がりました。
ただ、前回の最後で触れたように、API レスポンスから Address を生成する部分にはまだ課題が残っています。
今回の課題:API レスポンスとの直接結合
main.py で API レスポンスから Address インスタンスを生成している部分を見てみます。
address = Address(**response.json())
response.json() の戻り値(辞書)を直接 Address のコンストラクタに展開しています。
シンプルで良さそうに見えますが、実はこの書き方には罠があります。
1. 余分なフィールドでエラーになる
API のレスポンスに Address クラスが定義していないフィールドが含まれていたらどうなるか。たとえば latitude や longitude が追加されたとします。
payload = {
"zipcode": "1000001",
"prefecture": "東京都",
# ... 必要なフィールド ...
"latitude": 35.6762, # Address には存在しないフィールド
"longitude": 139.6503,
}
Address(**payload) # TypeError: 予期しないキーワード引数 'latitude'
dataclass のコンストラクタは定義されていないキーワード引数を受け取ると TypeError を発生させます。外部 API にフィールドが追加されただけでこちらのコードが壊れてしまう。正直これはかなり困ります。
2. 外部スキーマの変更がドメインモデルに直接影響する
もし将来的に、
- API 側で
prefectureがprefに変更されたら? -
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 なのか
Mapping は collections.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 オブジェクト(ChainMap や MappingProxyType など)で渡したくなっても、引数が Mapping なら受け入れられます。
具体的な実装ではなく「キーで値を読める」という振る舞いに依存する。これは 8 日目で紹介した Protocol と同じ考え方ですね。
MappingProxyType とは
MappingProxyType は types モジュールで提供される読み取り専用の 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())
得られるメリットは大きいです。
-
余計なフィールドを無視できる
-
payloadにlatitudeやlongitudeがあっても問題ありません。unmarshal_payloadで使うキーだけ取り出すのでTypeErrorは起きません。
-
-
型変換やバリデーションを書く場所ができる
- API が
zipcodeを整数で返すようになってもstr(payload["zipcode"])にしておけば呼び出し側は変更不要です。
- API が
-
外部スキーマの変更からドメインモデルを守れる
- API 側で
prefectureがprefに変わってもunmarshal_payloadの中を書き換えるだけで済みます。Addressを使っている他のコードには一切手を入れなくて済む。
- API 側で
unmarshal_payloadメソッド は「外部のゆるい世界」と「内部の厳密な世界」の境界として機能します。この境界があるかないかで保守性がまるで違います。
コレクション抽象型ファミリー
ここまで Mapping を使ってきましたが、collections.abc には同様の抽象型が他にもいくつか用意されています。せっかくなのでここで主要なものを整理しておきます。
主要なコレクション抽象型
| 型 | 説明 | 代表的な具象型 |
|---|---|---|
Mapping[K, V] |
キー → 値のマッピング(読み取り前提) |
dict(TypedDict も実体は dict) |
MutableMapping[K, V] |
キー → 値の変更可能マッピング | dict |
Sequence[T] |
インデックスアクセス可能(読み取り前提) |
list, tuple, str
|
MutableSequence[T] |
インデックスアクセス可能な変更可能コレクション | list |
Iterable[T] |
for ループで回せるもの全般 |
ほぼすべてのコレクション |
Collection[T] |
for で回せて len() が使えて in で含まれるか調べられる |
list, set, dict
|
使用例:Sequence と Iterable
Sequence と Iterable の使い分けも見ておきましょう。複数の住所を処理する関数を考えます。
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 を使う場面
Iterable は for で回せることだけを要求します。list、tuple、set、ジェネレータなど何でも受け取れます。一番緩い条件ですね。
# ジェネレータも OK
def address_generator() -> Iterable[Address]:
yield addr1
yield addr2
count_by_prefecture(address_generator())
collections.abc には他にもコレクション抽象型が用意されています。興味があれば公式ドキュメントを覗いてみてください。
コードの進化:スクリプトへの適用
それでは unmarshal_payload と Mapping をスクリプトに適用しましょう。
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