複数のWeb APIを共通の関数インターフェイスにまとめる方法－アダプタパターン入門

import os
import httpx

# 緯度・経度を入力して現在の天気情報を取得する
def get_weather_openweather(latitude: float, longitude: float) -> dict:
    api_key = os.environ["OWM_API_KEY"] # 環境変数からAPIキーを取得
    response = httpx.get(
        "https://api.openweathermap.org/data/2.5/weather", # エンドポイント
        params={ # クエリパラメータ
            "lat": latitude, # 緯度
            "lon": longitude, # 経度
            "appid": api_key, # APIキー
            "units": "metric", # 単位に摂氏を指定
        },
        timeout=10.0,
    )
    response.raise_for_status() # 通信エラーのチェック
    data = response.json() # レスポンスを辞書として取得
    return {
        "location": data["name"], # 場所
        "temperature_c": data["main"]["temp"], # 温度(℃)
        "description": data["weather"][0]["description"], # 気象状況
        "wind_kmph": data["wind"]["speed"] * 3.6,  # 風速（単位変換：m/s → km/h）
    }

result = get_weather_openweather(35.6895, 139.6917)
print(result)

{'location': 'Tokyo', 'temperature_c': 23.39, 'description': 'broken clouds', 'wind_kmph': 1.62}

import httpx

# 緯度・経度を入力して現在の天気情報を取得する
def get_weather_openmeteo(latitude: float, longitude: float) -> dict:
    response = httpx.get(
        "https://api.open-meteo.com/v1/forecast", # エンドポイント
        params={
            "latitude": latitude, # 緯度
            "longitude": longitude, # 経度
            "current": "temperature_2m,wind_speed_10m,weather_code", # 取得パラメータ
        },
        timeout=10.0,
    )
    response.raise_for_status() 
    data = response.json()
    current = data["current"]
    return {
        "location": f"({latitude}, {longitude})", # 緯度・経度
        "temperature_c": current["temperature_2m"], # 高度2mの温度（℃）
        "description": f"code {current['weather_code']}", # 天気コード（WMO weather interpretation codes）
        "wind_kmph": current["wind_speed_10m"], # 高度10mの風速 (km/h)
    }

result = get_weather_openmeteo(35.6895, 139.6917)
print(result)

{'location': '(35.6895, 139.6917)', 'temperature_c': 21.4, 'description': 'code 2', 'wind_kmph': 3.9}

if config.weather_service == "owm":
    weather = get_weather_openweather(35.6895, 139.6917)
elif config.weather_service == "om":
    weather = get_weather_openmeteo(35.6895, 139.6917)
else:
    raise ValueError(f"unknown service: {config.weather_service}")

project/
├── interface.py          # 共通インターフェイス（TypedDict）
├── weather.py            # call 関数、レジストリ
├── adapters/
│   ├── __init__.py       # 空でよい
│   ├── openweather.py    # OpenWeatherMap アダプタ
│   └── openmeteo.py      # Open-Meteo アダプタ
└── example.py            # 呼び出し側のサンプル

# interface.py
from typing import TypedDict

class WeatherInput(TypedDict):
   lat: float # 緯度
   lon: float # 経度

class WeatherOutput(TypedDict):
   location: str        # 場所名
   temperature_c: float # 温度（℃）
   description: str     # 天気の概要
   wind_kmph: float     # 風速（km/h）

# weather.py
from interface import WeatherInput, WeatherOutput

def call(service: str, input: WeatherInput) -> WeatherOutput:
   pass # 後で中身を追加

# adapters/openweather.py
import os
import httpx

from interface import WeatherInput, WeatherOutput

def fetch(input: WeatherInput) -> WeatherOutput:
    api_key = os.environ["OWM_API_KEY"]
    response = httpx.get(
        "https://api.openweathermap.org/data/2.5/weather",
        params={
            "lat": input["lat"],
            "lon": input["lon"],
            "appid": api_key,
            "units": "metric",
        },
        timeout=10.0,
    )
    response.raise_for_status()
    data = response.json()
    return {
        "location": data["name"],
        "temperature_c": data["main"]["temp"],
        "description": data["weather"][0]["description"],
        "wind_kmph": data["wind"]["speed"] * 3.6,  
    }

# adapters/openmeteo.py
import httpx

from interface import WeatherInput, WeatherOutput

# 天気コード（WMO weather interpretation codes）と天気概要の対応表
_WMO_CODE_TO_TEXT = {
    0:  "clear sky",
    1:  "mainly clear",
    2:  "partly cloudy",
    3:  "overcast",
    45: "fog",
    51: "light drizzle",
    61: "light rain",
    63: "moderate rain",
    71: "light snow",
    95: "thunderstorm",
}

def fetch(input: WeatherInput) -> WeatherOutput:
    response = httpx.get(
        "https://api.open-meteo.com/v1/forecast",
        params={
            "latitude": input["lat"],
            "longitude": input["lon"],
            "current": "temperature_2m,wind_speed_10m,weather_code",
        },
        timeout=10.0,
    )
    response.raise_for_status()
    data = response.json()
    current = data["current"]
    code = current["weather_code"]
    return {
        "location": f"({input['lat']}, {input['lon']})",
        "temperature_c": current["temperature_2m"],
        "description": _WMO_CODE_TO_TEXT.get(code, f"code {code}"), # 対応表による変換
        "wind_kmph": current["wind_speed_10m"],
    }

# weather.py
from collections.abc import Callable
from adapters import openweather, openmeteo
from interface import WeatherInput, WeatherOutput

# 関数のインターフェイスを定義（型ヒント）
WeatherAdapter = Callable[[WeatherInput], WeatherOutput]

_REGISTRY: dict[str, WeatherAdapter] = {  # レジストリ（辞書）の定義
    "ow": openweather.fetch,
    "om": openmeteo.fetch,
}

def call(service: str, input: WeatherInput) -> WeatherOutput:
    try:
        adapter = _REGISTRY[service]  # サービス名からアダプタを取得
    except KeyError:
        known = ", ".join(sorted(_REGISTRY))
        raise ValueError(f"unknown service: {service!r} (known: {known})")
    return adapter(input)  # アダプタ関数をcallして戻り値をそのまま返す

# example.py
from weather import call

result = call("ow", {"lat": 35.68, "lon": 139.69})
print(result)

result = call("om", {"lat": 35.68, "lon": 139.69})
print(result)

{'location': 'Tokyo', 'temperature_c': 17.48, 'description': 'overcast clouds', 'wind_kmph': 1.62}
{'location': '(35.68, 139.69)', 'temperature_c': 16.9, 'description': 'partly cloudy', 'wind_kmph': 1.6}

# example.py
import os

from weather import call

service = os.environ.get("WEATHER_SERVICE", "om")
result = call(service, {"lat": 35.68, "lon": 139.69})

# weather.py (レジストリへの追加部分のみ抽出）
from adapters import openweather, openmeteo, mock

_REGISTRY: dict[str, WeatherAdapter] = {
    "ow":   openweather.fetch,
    "om":   openmeteo.fetch,
    "mock": mock.fetch,
}

# adapters/mock.py
from interface import WeatherInput, WeatherOutput

def fetch(input: WeatherInput) -> WeatherOutput:
    return {
        "location": "mock",
        "temperature_c": 20.0,
        "description": "stub: always clear",
        "wind_kmph": 5.0,
    }

# weather.py（mock を加えたレジストリ）
from adapters import openmeteo, mock

_REGISTRY: dict[str, WeatherAdapter] = {
    "om":   openmeteo.fetch,
    "mock": mock.fetch,
}