A+B 複合 — 例外の握り潰し × null問題 × マジックナンバー（配送先住所検証処理 Bad→Good 第10章）× address-validation-service Geocoding API呼び出しの信頼性設計（リトライ判定 × Secret Manager × Workload Identity Federation × Cloud Tasksレート制御）

📋 コピー# tests/test_address_validation.py
import pytest
from address_validation import (
    GeocodedAddress, InvalidAddressError, ShippingAddress, validate_shipping_address,
)


class TestValidateShippingAddress:
    def test_raises_invalid_address_error_below_threshold(self, monkeypatch):
        monkeypatch.setattr(
            "address_validation.geocode_client.geocode",
            lambda **kw: FakeResult(confidence=0.2, lat=0, lng=0, normalized_address=""),
        )
        address = ShippingAddress("000-0000", "存在しない県", "", "")
        with pytest.raises(InvalidAddressError):
            validate_shipping_address(address)

    def test_wraps_timeout_as_service_unavailable(self, monkeypatch):
        def raise_timeout(**kw):
            raise GeocodeTimeoutError("timeout")
        monkeypatch.setattr("address_validation.geocode_client.geocode", raise_timeout)
        address = ShippingAddress("150-0002", "東京都", "渋谷区", "渋谷1-1-1")
        with pytest.raises(GeocodingServiceUnavailableError):
            validate_shipping_address(address)

    def test_does_not_return_none_on_failure(self, monkeypatch):
        # Noneを返す旧仕様への回帰防止テスト
        monkeypatch.setattr(
            "address_validation.geocode_client.geocode",
            lambda **kw: FakeResult(confidence=0.1, lat=0, lng=0, normalized_address=""),
        )
        address = ShippingAddress("000-0000", "存在しない県", "", "")
        try:
            result = validate_shipping_address(address)
            assert result is not None  # 到達しないはず(例外がraiseされる)
        except InvalidAddressError:
            pass  # 期待通り

📋 コピー"""geocoding_client.py — リトライ判定(修正2) × タイムアウト(修正1) × OTel計装(修正7)"""
from __future__ import annotations

import httpx
from opentelemetry import trace
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential_jitter,
)

from address_validation import (
    GeocodedAddress,
    GeocodingServiceUnavailableError,
    InvalidAddressError,
    ShippingAddress,
)

tracer = trace.get_tracer(__name__)

# 修正1: 接続/読み取りタイムアウトを明示的に分けて設定
_HTTP_TIMEOUT = httpx.Timeout(connect=2.0, read=5.0, write=2.0, pool=2.0)
_http_client = httpx.Client(timeout=_HTTP_TIMEOUT)


@retry(
    # 修正2: 一時的エラー(GeocodingServiceUnavailableError)のみリトライ対象にする。
    # InvalidAddressError(恒久的エラー)はリトライせず即座に呼び出し元へ伝播する。
    retry=retry_if_exception_type(GeocodingServiceUnavailableError),
    wait=wait_exponential_jitter(initial=1, max=10),  # 修正3: 指数バックオフ+ジッター
    stop=stop_after_attempt(3),
    reraise=True,
)
def call_geocoding_api(address: ShippingAddress, api_key: str) -> GeocodedAddress:
    """外部Geocoding APIを呼び出し、住所を緯度経度へ変換する。

    Raises:
        InvalidAddressError: 住所自体が不正な場合(リトライしない)。
        GeocodingServiceUnavailableError: 3回リトライしても解決しなかった場合。
    """
    with tracer.start_as_current_span("geocode_api_call") as span:  # 修正7
        span.set_attribute("address.postal_code", address.postal_code)
        try:
            response = _http_client.get(
                "https://geocoding.example.com/v1/lookup",
                params={
                    "postal_code": address.postal_code,
                    "prefecture": address.prefecture,
                    "city": address.city,
                    "street": address.street,
                },
                headers={"Authorization": f"Bearer {api_key}"},
            )
        except httpx.TimeoutException as exc:
            span.set_attribute("error.type", "timeout")  # 修正7: エラー種別を計装
            raise GeocodingServiceUnavailableError("タイムアウト") from exc

        if response.status_code == 429:
            span.set_attribute("error.type", "quota_exceeded")
            raise GeocodingServiceUnavailableError("クォータ超過") from None
        if response.status_code >= 500:
            span.set_attribute("error.type", "server_error")
            raise GeocodingServiceUnavailableError(
                f"Geocoding APIの内部エラー: {response.status_code}"
            ) from None
        if response.status_code >= 400:
            span.set_attribute("error.type", "invalid_address")
            raise InvalidAddressError(address, confidence=0.0) from None

        payload = response.json()
        span.set_attribute("error.type", "none")
        return GeocodedAddress(
            lat=payload["lat"],
            lng=payload["lng"],
            normalized_address=payload["normalized_address"],
        )

Bad vs Good 設計比較

確認コマンド

📋 コピー# 1. Secret ManagerにAPIキーが登録されていることを確認
gcloud secrets versions list geocoding-api-key --format="value(name,state)"
# Expected: 1  ENABLED

# 2. Cloud RunサービスアカウントがSecret Accessorロールを持つことを確認
gcloud secrets get-iam-policy geocoding-api-key \
  --format="value(bindings.role)" --flatten="bindings[]"
# Expected: roles/secretmanager.secretAccessor が含まれる

# 3. Workload Identity Federationでの認証が機能することを確認(GitHub Actions実行ログ)
gh run view --log | grep "Successfully authenticated"
# Expected: Successfully authenticated via Workload Identity Federation

# 4. Cloud Tasksキューのレート制限を確認
gcloud tasks queues describe geocoding-validation-queue --location=asia-northeast1 \
  --format="value(rateLimits.maxDispatchesPerSecond)"
# Expected: 10.0

# 5. 恒久的エラーがリトライされないことを確認(ユニットテスト)
python -c "
from geocoding_client import call_geocoding_api
from address_validation import InvalidAddressError, ShippingAddress
try:
    call_geocoding_api(ShippingAddress('000-0000','存在しない県','',''), api_key='dummy')
except InvalidAddressError:
    print('リトライされず即座に伝播: OK')
"