Recustomer Tech BlogPublicationへの投稿
🐍

Pythonの型ヒントと共に進化するコード（#22: Literal と Annotated）

K'
に公開
2025/12/29
Python
アドベントカレンダー
Mypy
型ヒント
tech
これまでの連載記事


前回は Result 型を導入し、成功と失敗という 2 つの道を統一された型で扱えるようになりました。

しかし、ここで一度コード全体を見返してみましょう。

今回の課題：print のままのログ出力

main.py の handle_fetch_error 関数
def handle_fetch_error(error: FetchError) -> None:
    match error.type:
        case FetchErrorType.NETWORK_ERROR:
            print(f"ネットワークエラー: {error.message}")
        case FetchErrorType.NOT_FOUND:
            print(f"郵便番号が見つかりません: {error.message}")
        # ...

エラーの型は整理されました。処理の分岐も網羅性が保証されています。しかし、肝心の出力部分は print のままです。

処方箋：print から logging へ

print は手軽ですが、ログレベルの制御や出力先の切り替えができません。本格的なアプリケーションでは Python 標準の logging モジュールを使ったほうがいいです。

import logging

logger = logging.getLogger(__name__)
logger.info("処理を開始")
logger.error("エラー発生")

logging では 5 段階のログレベル（DEBUG/INFO/WARNING/ERROR/CRITICAL）で出力を制御でき、出力先やフォーマットも柔軟に設定できます。

取りうる値を型で制限する Literal

ここで、ログレベルを設定する関数を書くことを考えてみましょう。

def setup_logger(name: str, level: str) -> logging.Logger:
    logger = logging.getLogger(name)
    logger.setLevel(level)
    # ...

level: str という型注釈では、どんな文字列でも受け入れてしまいます。"INFO" と書くべきところを "INOF" とタイポしても型チェッカーは何も警告してくれません。

こうした文字列だが取りうる値は決まっているというケースには Literal が最適です。

from typing import Literal

# ログレベルを型で制限
LogLevel = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]

def setup_logger(name: str, level: LogLevel) -> logging.Logger:
    logger = logging.getLogger(name)
    logger.setLevel(level)
    # ...

Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] は、これら 5 つの文字列リテラルのいずれかだけを許容する型です。

setup_logger("app", "INFO")      # OK
setup_logger("app", "DEBUG")     # OK
setup_logger("app", "INOF")      # 型エラー！ "INOF" は LogLevel ではない
setup_logger("app", "verbose")   # 型エラー！

Literal は Union の一種を具体的な値レベルまで落とし込んだものだと捉えることができます。設定値やフラグなど、取りうるパターンが限られている場面で綴りミスや無効な指定を早い段階で防げます。

型にメタデータを付与する Annotated

Literal でログレベルを型安全にできました。では、ロガー名についてはどうでしょうか。

def setup_logger(name: str, level: LogLevel) -> logging.Logger:
    # ...

name: str という型注釈は正確ですが、この引数が何を意図しているのかは型からは読み取れません。モジュール名を渡すべき、空文字列は避けるべきといった情報はドキュメントに頼るしかありません。

6 日目で紹介した NewType を使えば LoggerName = NewType("LoggerName", str) のように意味を持たせた型を作れます。「空文字列を弾く」といった制約を厳密に扱うなら専用の型やバリデーション関数を用意する方法もあります。

ここでは別のアプローチとして Annotated を紹介します。

Annotated を使うと、型そのものにメタデータを付与できます。

from typing import Annotated

# 型にメタデータ（説明）を付与
LoggerName = Annotated[str, "モジュール名として使用されるロガー名"]

def setup_logger(name: LoggerName, level: LogLevel) -> logging.Logger:
    # ...

Annotated[str, "モジュール名として使用されるロガー名"] において、型チェッカーは第 1 引数の str だけを見ます。

Annotated の仕組みを覗いてみる

Annotated が実際にどう機能するのか、Pydantic などのライブラリを使わずに「空文字列を弾く」ための自作バリデーターを書いて確かめてみましょう。

Annotated の第 2 引数以降には何でも置けます。文字列でも、オブジェクトでも、関数でも構いません。型チェッカーはこれらを無視しますが、実行時には __metadata__ 属性としてアクセスできます。

この仕組みを利用してバリデーション関数を型に埋め込んでみます。

まずは、埋め込み対象となるバリデーション関数を定義します。

from dataclasses import dataclass
from typing import Annotated

# Result 型は21日目で実装したものを使用
type Result[T, E] = Ok[T] | Err[E]

@dataclass(frozen=True, slots=True)
class EmptyStringError:
    """空文字列エラー"""
    message: str = "空文字列は許可されていません"

# バリデーション関数
def non_empty(value: str) -> Result[str, EmptyStringError]:
    """空文字列を許可しない"""
    if not value or value.isspace():
        return Err(EmptyStringError())
    return Ok(value)

21 日目で実装した Result 型を使って失敗の可能性を型で表現しています。

次に、このメタデータを読み取ってバリデーションを実行するデコレータを書きます。

import inspect
from dataclasses import dataclass
from functools import wraps
from typing import Annotated, get_args, get_origin, get_type_hints, Callable, Any

@dataclass(frozen=True, slots=True)
class ValidationError:
    """バリデーション失敗時のエラー"""
    field: str
    message: str

def validate_args[**P, R](
    func: Callable[P, Result[R, ValidationError]]
) -> Callable[P, Result[R, ValidationError]]:
    """Annotated のメタデータに callable があれば適用し、Result で返す"""
    # 👉 include_extras=True で Annotated のメタデータを保持
    hints = get_type_hints(func, include_extras=True)
    sig = inspect.signature(func)

    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> Result[R, ValidationError]:
        # 👉 位置引数とキーワード引数を統一的に扱う
        bound = sig.bind(*args, **kwargs)
        bound.apply_defaults()

        for name, value in bound.arguments.items():
            hint = hints.get(name)
            # 👉 get_origin で Annotated かどうかを判定
            if hint is not None and get_origin(hint) is Annotated:
                # 👉 get_args で (基底型, *メタデータ) を取得
                type_args = get_args(hint)
                for meta in type_args[1:]:
                    # 👉 メタデータが callable ならバリデーション関数として実行
                    if callable(meta):
                        result = meta(value)
                        if is_err(result):
                            return Err(
                                ValidationError(field=name, message=result.error.message)
                            )
                        value = result.value
            bound.arguments[name] = value

        # 👉 元の引数構造を保持したまま呼び出す
        return func(*bound.args, **bound.kwargs)

    return wrapper

def validate_args[**P, R](...) という構文は、16 日目や17 日目で紹介した Generics の記法です。**P で関数のパラメータ全体を型として保持し、R で戻り値の型を表します。

inspect.signature で関数のシグネチャを取得し、bind で位置引数とキーワード引数を統一的に束縛します。get_origin と get_args で Annotated 型を判定・分解するのが標準的な方法です。

では、このデコレータを使ってみましょう。
先ほど定義した non_empty を Annotated のメタデータとして埋め込んだ型 LoggerName を作ります。デコレータはこの型から non_empty を取り出して実行時にバリデーションを適用します。

# 👉 Annotated でバリデーション関数を型に埋め込む
LoggerName = Annotated[str, non_empty]
LogLevel = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]

# 👉 Annotated でバリデーション関数を型に埋め込む
@validate_args
def setup_logger(
    name: LoggerName,
    level: LogLevel = "INFO",
) -> Result[logging.Logger, ValidationError]:
    """ロガーを設定する"""
    logger = logging.getLogger(name)
    logger.setLevel(level)
    return Ok(logger)

# 正常系
result = setup_logger(name="app", level="INFO")
if is_err(result):
    print(f"Error: {result.error}")
else:
    logger = result.value
    logger.info("Logger initialized successfully")

# エラー系（空文字列）
result = setup_logger(name="", level="DEBUG")
if is_err(result):
    print(f"Error: field={result.error.field}, message={result.error.message}")
    # → Error: field=name, message=空文字列は許可されていません

Pydantic や FastAPI がやっていることの本質はこれです。Annotated のメタデータを読み取って処理しているだけなんですよね。

実際に以下のようなコードを書いたことがある人も多いのではないでしょうか。

# Pydantic の例
from pydantic import BaseModel, Field

class User(BaseModel):
    name: Annotated[str, Field(min_length=1, max_length=100)]
    age: Annotated[int, Field(ge=0, le=150)]

# FastAPI の例
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(
    q: Annotated[str | None, Query(max_length=50)] = None
):
    ...

これらはすべて Annotated の __metadata__ からバリデーションルールやパラメータ情報を取り出して処理しています。仕組みを理解しておくとなぜ Annotated を使うのかが腑に落ちるはずです。

コードの進化：print から logging へ

それでは、これらの型ツールを使って print を logging に置き換えましょう。

変更前の状態

# main.py
def handle_fetch_error(error: FetchError) -> None:
    match error.type:
        case FetchErrorType.NETWORK_ERROR:
            print(f"ネットワークエラー: {error.message}")
        # ...

logging の適用

まず、先ほど作成したバリデーターを validator.py として保存します。

validator.py (新規作成)

# validator.py
from __future__ import annotations

import inspect
from dataclasses import dataclass
from functools import wraps
from typing import Annotated, Any, Callable, get_args, get_origin, get_type_hints

from result import Err, Ok, Result, is_err


@dataclass(frozen=True, slots=True)
class EmptyStringError:
    """空文字列エラー"""

    message: str = "空文字列は許可されていません"


@dataclass(frozen=True, slots=True)
class ValidationError:
    """バリデーション失敗時のエラー"""

    field: str
    message: str


def non_empty(value: str) -> Result[str, EmptyStringError]:
    """空文字列を許可しない"""
    if not value or value.isspace():
        return Err(EmptyStringError())
    return Ok(value)


def validate_args[**P, R](
    func: Callable[P, Result[R, ValidationError]],
) -> Callable[P, Result[R, ValidationError]]:
    """Annotated のメタデータに callable があれば適用し、Result で返す"""
    hints = get_type_hints(func, include_extras=True)
    sig = inspect.signature(func)

    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> Result[R, ValidationError]:
        bound = sig.bind(*args, **kwargs)
        bound.apply_defaults()

        for name, value in bound.arguments.items():
            hint = hints.get(name)
            if hint is not None and get_origin(hint) is Annotated:
                type_args = get_args(hint)
                for meta in type_args[1:]:
                    if callable(meta):
                        result = meta(value)
                        if is_err(result):
                            return Err(
                                ValidationError(field=name, message=result.error.message)
                            )
                        value = result.value
            bound.arguments[name] = value

        return func(*bound.args, **bound.kwargs)

    return wrapper

次に、このバリデーターを使って logger.py を作成します。

logger.py (新規作成)

# logger.py
from __future__ import annotations

import logging
from typing import Annotated, Literal

from result import Result, Ok
from validator import ValidationError, non_empty, validate_args

# Literal でログレベルを型安全に
LogLevel = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]

# Annotated でバリデーション関数を付与
LoggerName = Annotated[str, non_empty]

@validate_args
def setup_logger(
    name: LoggerName,
    level: LogLevel = "INFO",
) -> Result[logging.Logger, ValidationError]:
    """指定された名前とレベルでロガーを設定する"""
    logger = logging.getLogger(name)
    logger.setLevel(level)
    # 👉 ハンドラの追加は呼び出し側に委ねる
    return Ok(logger)

main.py (変更後)

# main.py
from __future__ import annotations

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

from models import (
    ZipCode,
    Headers,
    Address,
    AddressFormatter,
)
from http_client import (
    HttpClient,
    RequestsHttpClient,
    FetchError,
    FetchErrorType,
    to_error_type,
)
from result import Result, Ok, Err, is_err, is_ok
from logger import setup_logger  # 👉 ロガーをインポート

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

# 👉 モジュールレベルでロガーを取得（Result 型で返ってくる）
logger_result = setup_logger(name=__name__)
if is_err(logger_result):
    # ロガーの初期化に失敗した場合はプログラムを終了
    raise RuntimeError(f"Logger initialization failed: {logger_result.error}")
logger = logger_result.value

def handle_fetch_error(error: FetchError) -> None:
    """エラーの種類に応じてログを出力する"""
    match error.type:
        case FetchErrorType.NETWORK_ERROR:
            # 👉 print → logger.error
            logger.error("ネットワークエラー: %s", error.message)
        case FetchErrorType.NOT_FOUND:
            logger.warning("郵便番号が見つかりません: %s", error.message)
        case FetchErrorType.CLIENT_ERROR:
            logger.error("リクエストエラー: %s", error.message)
        case FetchErrorType.SERVER_ERROR:
            logger.error("サーバーエラー: %s", error.message)
        case FetchErrorType.API_ERROR:
            logger.error("APIエラー: %s", error.message)
        case _:
            assert_never(error.type)

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

    api_url = f"{BASE_URL}{Address.API_PATH}"
    # 👉 処理開始をログ出力
    logger.info("住所検索を開始: zipcode=%s", zipcode)

    try:
        response = http_client.post(api_url, json={"zipcode": zipcode}, headers=headers)
        if response.status_code != HTTP_OK:
            error_type = to_error_type(response.status_code)
            # 👉 エラー時のログ
            logger.warning(
                "API がエラーを返却: status=%d, zipcode=%s",
                response.status_code,
                zipcode,
            )
            return Err(FetchError(
                type=error_type,
                message=f"Address fetch failed with status {response.status_code}",
            ))

        payload = cast(Mapping[str, Any], response.json())
        address = Address.unmarshal_payload(payload)
        formatter = AddressFormatter()
        result = formatter.with_address(address).with_kana(include_kana).build()

        # 👉 成功時のログ
        logger.info("住所検索が完了: zipcode=%s", zipcode)
        return Ok(json.dumps(result, indent=2, ensure_ascii=False))

    except Exception as e:
        logger.exception("予期せぬエラーが発生")
        return Err(FetchError(type=FetchErrorType.NETWORK_ERROR, message=str(e)))

# 実行例
if __name__ == "__main__":
    # 👉 アプリケーション側でログ出力の設定を行う
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    )

    http_client = RequestsHttpClient()
    zipcode = ZipCode("1000001")
    result = fetch_and_format_address(
        zipcode, include_kana=True, http_client=http_client
    )

    if is_err(result):
        handle_fetch_error(result.error)
    else:
        # 👉 result.value は既に JSON 文字列なのでそのまま出力
        print(result.value)

得られたもの

今回は print を logging に置き換え、ログ出力を型安全に整備しました。

Literal でログレベルをタイポから守り、Annotated で型に意図を埋め込む。正しい値だけを受け入れるという型ヒントの本質的な価値をこの 2 つがよく体現しています。

次回予告

連載を通じて磨き上げてきたこのコードは、使い捨てではなく再利用されることを前提としたライブラリとしての形が見えてきました。

せっかくなら、これをパッケージとして配布して他のプロジェクトからもそのまま使えるようにしたいところです。
ただ、コードをそのままパッケージ化しただけでは型情報は利用者側に正しく伝わりません。

第三者がインストールして利用すると、すべての関数が Any 扱いになり補完も効かなくなります。それは避けたいところです。

次回は型情報をパッケージの一部として正しく届けるための仕上げを行います。

👉 23 日目: py.typed と型情報の公開

処方後のコードはこちら

validator.py（新規作成）

# validator.py
from __future__ import annotations

import inspect
from dataclasses import dataclass
from functools import wraps
from typing import Annotated, Any, Callable, get_args, get_origin, get_type_hints

from result import Err, Ok, Result, is_err


@dataclass(frozen=True, slots=True)
class EmptyStringError:
    """空文字列エラー"""

    message: str = "空文字列は許可されていません"


@dataclass(frozen=True, slots=True)
class ValidationError:
    """バリデーション失敗時のエラー"""

    field: str
    message: str


def non_empty(value: str) -> Result[str, EmptyStringError]:
    """空文字列を許可しない"""
    if not value or value.isspace():
        return Err(EmptyStringError())
    return Ok(value)


def validate_args[**P, R](
    func: Callable[P, Result[R, ValidationError]],
) -> Callable[P, Result[R, ValidationError]]:
    """Annotated のメタデータに callable があれば適用し、Result で返す"""
    hints = get_type_hints(func, include_extras=True)
    sig = inspect.signature(func)

    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> Result[R, ValidationError]:
        bound = sig.bind(*args, **kwargs)
        bound.apply_defaults()

        for name, value in bound.arguments.items():
            hint = hints.get(name)
            if hint is not None and get_origin(hint) is Annotated:
                type_args = get_args(hint)
                for meta in type_args[1:]:
                    if callable(meta):
                        result = meta(value)
                        if is_err(result):
                            return Err(
                                ValidationError(field=name, message=result.error.message)
                            )
                        value = result.value
            bound.arguments[name] = value

        return func(*bound.args, **bound.kwargs)

    return wrapper

logger.py（新規作成）

# logger.py
from __future__ import annotations

import logging
from typing import Annotated, Literal

from result import Result, Ok
from validator import ValidationError, non_empty, validate_args

# Literal でログレベルを型安全に
LogLevel = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]

# Annotated でバリデーション関数を付与
LoggerName = Annotated[str, non_empty]

@validate_args
def setup_logger(
    name: LoggerName,
    level: LogLevel = "INFO",
) -> Result[logging.Logger, ValidationError]:
    """指定された名前とレベルでロガーを設定する"""
    logger = logging.getLogger(name)
    logger.setLevel(level)
    # ハンドラの追加は呼び出し側に委ねる
    return Ok(logger)

result.py 

# result.py
from __future__ import annotations

from dataclasses import dataclass
from typing import TypeIs, final  # TypeIs は 3.13 以降


@final
@dataclass(frozen=True, slots=True)
class Ok[T]:
    """成功を表すコンテナ"""
    value: T


@final
@dataclass(frozen=True, slots=True)
class Err[E]:
    """失敗を表すコンテナ"""
    error: E


type Result[T, E] = Ok[T] | Err[E]


def is_ok[T, E](result: Result[T, E]) -> TypeIs[Ok[T]]:
    """成功かどうかを判定し、型を絞り込む"""
    return isinstance(result, Ok)


def is_err[T, E](result: Result[T, E]) -> TypeIs[Err[E]]:
    """失敗かどうかを判定し、型を絞り込む"""
    return isinstance(result, Err)

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

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

@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"]),
        )

@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"]),
        )

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

type ApiResponse = Address | ApiError

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

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

typings.py 

# typings.py
from __future__ import annotations

from collections.abc import Mapping, Sequence


def first[T](items: Sequence[T]) -> T | None:
    """シーケンスの最初の要素を返す。空なら None を返す。"""
    return items[0] if items else None


def get_or[K, V](d: Mapping[K, V], key: K, default: V) -> V:
    """マッピングから値を取得する。キーがなければデフォルト値を返す。"""
    return d.get(key, default)

http_client.py 

# http_client.py
from __future__ import annotations

import requests as requests_lib
from dataclasses import dataclass
from enum import Enum, auto
from typing import Protocol

from decorators import measure_time
from models import Headers

type JsonObject = dict[str, object]

class FetchErrorType(Enum):
    NETWORK_ERROR = auto()
    NOT_FOUND = auto()
    CLIENT_ERROR = auto()
    SERVER_ERROR = auto()
    API_ERROR = auto()

@dataclass(frozen=True, slots=True)
class FetchError:
    type: FetchErrorType
    message: str


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)


def to_error_type(status_code: int) -> FetchErrorType:
    """HTTPステータスコードをFetchErrorTypeに変換する"""
    match status_code:
        case 404:
            return FetchErrorType.NOT_FOUND
        case num if 400 <= num < 500:
            return FetchErrorType.CLIENT_ERROR
        case _:
            return FetchErrorType.SERVER_ERROR

main.py 

# main.py
from __future__ import annotations

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

from models import (
    ZipCode,
    Headers,
    Address,
    AddressFormatter,
)
from http_client import (
    HttpClient,
    RequestsHttpClient,
    FetchError,
    FetchErrorType,
    to_error_type,
)
from result import Result, Ok, Err, is_err, is_ok
from logger import setup_logger

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

# モジュールレベルでロガーを取得（Result 型で返ってくる）
logger_result = setup_logger(name=__name__)
if is_err(logger_result):
    # ロガーの初期化に失敗した場合はプログラムを終了
    raise RuntimeError(f"Logger initialization failed: {logger_result.error}")
logger = logger_result.value

def handle_fetch_error(error: FetchError) -> None:
    """エラーの種類に応じてログを出力する"""
    match error.type:
        case FetchErrorType.NETWORK_ERROR:
            logger.error("ネットワークエラー: %s", error.message)
        case FetchErrorType.NOT_FOUND:
            logger.warning("郵便番号が見つかりません: %s", error.message)
        case FetchErrorType.CLIENT_ERROR:
            logger.error("リクエストエラー: %s", error.message)
        case FetchErrorType.SERVER_ERROR:
            logger.error("サーバーエラー: %s", error.message)
        case FetchErrorType.API_ERROR:
            logger.error("APIエラー: %s", error.message)
        case _:
            assert_never(error.type)

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

    api_url = f"{BASE_URL}{Address.API_PATH}"
    logger.info("住所検索を開始: zipcode=%s", zipcode)

    try:
        response = http_client.post(api_url, json={"zipcode": zipcode}, headers=headers)
        if response.status_code != HTTP_OK:
            error_type = to_error_type(response.status_code)
            logger.warning(
                "API がエラーを返却: status=%d, zipcode=%s",
                response.status_code,
                zipcode,
            )
            return Err(FetchError(
                type=error_type,
                message=f"Address fetch failed with status {response.status_code}",
            ))

        payload = cast(Mapping[str, Any], response.json())
        address = Address.unmarshal_payload(payload)
        formatter = AddressFormatter()
        result = formatter.with_address(address).with_kana(include_kana).build()

        logger.info("住所検索が完了: zipcode=%s", zipcode)
        return Ok(json.dumps(result, indent=2, ensure_ascii=False))

    except Exception as e:
        logger.exception("予期せぬエラーが発生")
        return Err(FetchError(type=FetchErrorType.NETWORK_ERROR, message=str(e)))

# 実行例
if __name__ == "__main__":
    # アプリケーション側でログ出力の設定を行う
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    )

    http_client = RequestsHttpClient()
    zipcode = ZipCode("1000001")
    result = fetch_and_format_address(
        zipcode, include_kana=True, http_client=http_client
    )

    if is_err(result):
        handle_fetch_error(result.error)
    else:
        print(result.value)
Recustomer Tech Blog
Recustomer Tech BlogPublication

Discussion