Python FastAPIで構築する実用的データ統合パイプライン: 天気・交通APIを例にしたジェネレータ活用術

はじめに

こんにちは、皆さん。今回は、FastAPIを使用して天気予報APIと交通情報APIを作成し、それらから取得したデータを効率的に統合する方法について、Pythonのジェネレータを使用したアプローチを紹介します。この例を通じて、複数のデータソースを組み合わせることで、日常生活の計画をより効率的に立てる方法を学びましょう。

目次

1. はじめに
2. 環境準備
3. FastAPIを使用した天気・交通APIの作成
4. ジェネレータを使用したデータ統合
5. 動作確認と日常生活への応用
6. パフォーマンスと拡張性
7. まとめ

はじめに

日々の生活において、天気予報や交通情報は私たちの行動計画に大きな影響を与えます。本記事では、これらの情報を提供する2つのAPIを作成し、それらからのデータを効率的に統合して活用する方法を紹介します。

環境準備

まず、必要なライブラリをインストールします：

pip install fastapi uvicorn requests

FastAPIを使用した天気・交通APIの作成

api_server.pyファイルを作成し、以下のコードを記述します：

from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn
from typing import List
import random
from datetime import datetime, timedelta

app = FastAPI()

class WeatherForecast(BaseModel):
    date: str
    temperature: float
    condition: str

class TrafficInfo(BaseModel):
    date: str
    congestion_level: str
    delay_minutes: int

@app.get("/api/weather", response_model=List[WeatherForecast])
async def get_weather_forecast():
    conditions = ["晴れ", "曇り", "雨", "雪"]
    forecasts = []
    for i in range(7):  # 1週間分の天気予報
        date = (datetime.now() + timedelta(days=i)).strftime("%Y-%m-%d")
        forecasts.append(WeatherForecast(
            date=date,
            temperature=round(random.uniform(0, 35), 1),
            condition=random.choice(conditions)
        ))
    return forecasts

@app.get("/api/traffic", response_model=List[TrafficInfo])
async def get_traffic_info():
    congestion_levels = ["低", "中", "高"]
    traffic_info = []
    for i in range(7):  # 1週間分の交通情報
        date = (datetime.now() + timedelta(days=i)).strftime("%Y-%m-%d")
        traffic_info.append(TrafficInfo(
            date=date,
            congestion_level=random.choice(congestion_levels),
            delay_minutes=random.randint(0, 60)
        ))
    return traffic_info

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

このコードは、天気予報（/api/weather）と交通情報（/api/traffic）を提供する2つのエンドポイントを持つFastAPIアプリケーションを作成します。各エンドポイントは1週間分のデータをランダムに生成して返します。

ジェネレータを使用したデータ統合

次に、これらのAPIからデータを取得し統合するコードを作成します。data_integrator.pyファイルを作成し、以下のコードを記述します：

import requests
from datetime import datetime

def fetch_data_from_api(endpoint):
    response = requests.get(f"http://localhost:8000{endpoint}")
    return response.json()

def data_integrator():
    weather_data = fetch_data_from_api("/api/weather")
    traffic_data = fetch_data_from_api("/api/traffic")
    
    for weather, traffic in zip(weather_data, traffic_data):
        yield {
            "date": weather["date"],
            "weather": weather,
            "traffic": traffic
        }

def plan_week():
    today = datetime.now().date()
    for day_data in data_integrator():
        date = datetime.strptime(day_data["date"], "%Y-%m-%d").date()
        weather = day_data["weather"]
        traffic = day_data["traffic"]
        
        if date == today:
            continue  # 今日のデータはスキップ
        
        plan = f"日付: {date}\n"
        plan += f"天気: {weather['condition']}、気温: {weather['temperature']}℃\n"
        plan += f"交通: 混雑度 {traffic['congestion_level']}、予想遅延 {traffic['delay_minutes']}分\n"
        
        if weather["condition"] == "雨" and traffic["congestion_level"] == "高":
            plan += "行動プラン: 在宅勤務をおすすめします。\n"
        elif weather["condition"] in ["晴れ", "曇り"] and traffic["congestion_level"] == "低":
            plan += "行動プラン: オフィス勤務に最適な日です。\n"
        else:
            plan += "行動プラン: 通常通りの予定で問題ありません。\n"
        
        yield plan

if __name__ == "__main__":
    print("来週の行動計画:")
    for daily_plan in plan_week():
        print(daily_plan)
        print("-" * 40)

このdata_integrator関数は、天気予報APIと交通情報APIからのデータを統合し、日付ごとにまとめて生成します。plan_week関数は、統合されたデータを基に、各日の行動計画を生成します。

動作確認と日常生活への応用

1. まず、APIサーバーを起動します：

python api_server.py

1. 別のターミナルで、データ統合と行動計画生成スクリプトを実行します：

python data_integrator.py

このスクリプトは、1週間分の天気予報と交通情報を取得し、それらを基に各日の行動計画を生成して表示します。

出力例：

来週の行動計画:
日付: 2024-09-02
天気: 晴れ、気温: 28.5℃
交通: 混雑度 低、予想遅延 15分
行動プラン: オフィス勤務に最適な日です。
----------------------------------------
日付: 2024-09-03
天気: 雨、気温: 22.3℃
交通: 混雑度 高、予想遅延 45分
行動プラン: 在宅勤務をおすすめします。
----------------------------------------
...

この例では、天気と交通状況に基づいて、オフィス勤務か在宅勤務かを提案しています。実際の使用では、より複雑な条件や個人の予定を考慮に入れることで、より適切な行動計画を立てることができます。

パフォーマンスと拡張性

ジェネレータを使用することで、以下のような利点があります：

1. メモリ効率：1週間分のデータを一度にメモリに読み込む代わりに、1日ずつ処理できます。これにより、大量のデータを扱う場合でもメモリ使用量を抑えることができます。

2. リアルタイム処理：新しいデータが利用可能になった時点で即座に処理を開始できます。例えば、天気予報が更新されたら、すぐにその情報を反映した行動計画を生成できます。

3. 柔軟性：新しいデータソースを簡単に追加できます。これは、システムの拡張性を高める重要な特徴です。

システムの拡張例

例えば、より豊かな行動計画を立てるために、地域のイベント情報を追加したいと考えたとします。この場合、以下のようにシステムを拡張できます：

1. まず、api_server.py にイベント情報を提供する新しいエンドポイントを追加します：

class EventInfo(BaseModel):
    date: str
    event_name: str
    location: str

@app.get("/api/events", response_model=List[EventInfo])
async def get_event_info():
    events = []
    for i in range(7):  # 1週間分のイベント情報
        date = (datetime.now() + timedelta(days=i)).strftime("%Y-%m-%d")
        events.append(EventInfo(
            date=date,
            event_name=f"イベント{i+1}",
            location=f"場所{i+1}"
        ))
    return events

1. 次に、data_integrator.py の data_integrator 関数を拡張して、新しいAPIからのデータを統合します：

def data_integrator():
    weather_data = fetch_data_from_api("/api/weather")
    traffic_data = fetch_data_from_api("/api/traffic")
    event_data = fetch_data_from_api("/api/events")
    
    for weather, traffic, event in zip(weather_data, traffic_data, event_data):
        yield {
            "date": weather["date"],
            "weather": weather,
            "traffic": traffic,
            "event": event
        }

1. 最後に、plan_week 関数を更新して、イベント情報を考慮した行動計画を生成します：

def plan_week():
    today = datetime.now().date()
    for day_data in data_integrator():
        date = datetime.strptime(day_data["date"], "%Y-%m-%d").date()
        weather = day_data["weather"]
        traffic = day_data["traffic"]
        event = day_data["event"]
        
        if date == today:
            continue  # 今日のデータはスキップ
        
        plan = f"日付: {date}\n"
        plan += f"天気: {weather['condition']}、気温: {weather['temperature']}℃\n"
        plan += f"交通: 混雑度 {traffic['congestion_level']}、予想遅延 {traffic['delay_minutes']}分\n"
        plan += f"イベント: {event['event_name']} @ {event['location']}\n"
        
        if weather["condition"] == "雨" and traffic["congestion_level"] == "高":
            plan += "行動プラン: 在宅勤務をおすすめします。\n"
        elif weather["condition"] in ["晴れ", "曇り"] and traffic["congestion_level"] == "低":
            if event["event_name"] != "イベントなし":
                plan += f"行動プラン: オフィス勤務に最適な日です。仕事後に{event['event_name']}に参加するのはいかがでしょうか？\n"
            else:
                plan += "行動プラン: オフィス勤務に最適な日です。\n"
        else:
            plan += "行動プラン: 通常通りの予定で問題ありません。\n"
        
        yield plan

この拡張により、天気、交通、そしてイベント情報を総合的に考慮した、より詳細な行動計画を生成できるようになります。ジェネレータを使用しているため、新しいデータソースを追加しても、メモリ効率や処理の柔軟性は維持されます。

このような拡張性は、実際のプロジェクトにおいて非常に重要です。ユーザーのニーズの変化や、新しいデータソースの登場に応じて、システムを柔軟に進化させることができます。

まとめ

この記事では、FastAPIを使用して天気予報と交通情報のAPIを作成し、それらのデータをPythonのジェネレータを使用して効率的に統合する方法を学びました。この手法を応用することで、日々の生活における意思決定をサポートする実用的なツールを作成できます。

実際のプロジェクトでは、より正確なデータソースの使用や、ユーザーの個人的な予定やプリファレンスの考慮など、さらなる改良が必要になるでしょう。しかし、ここで示した基本的な概念は、効率的なデータ処理と意思決定支援システムの強力な基盤となります。

ぜひ、この例を基に、自分の生活に役立つデータ統合ツールを作ってみてください。API統合とジェネレータの力を活用することで、日常生活をより効率的に、そしてスマートにすることができるはずです。

Happy coding and smart living!