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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/e747ff45-92e5-1cfd-03bb-43bb7316183a.png)

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

## はじめに

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

## 環境準備

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

```bash
pip install fastapi uvicorn requests
```

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/51c134c9-58a5-7f7c-7b77-857abd2e87ce.png)

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

```python
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週間分のデータをランダムに生成して返します。

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/a2babaf0-5528-87c0-ace6-9bb402ff74e9.png)

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

```mermaid
sequenceDiagram
    participant Main
    participant plan_week
    participant data_integrator
    participant fetch_data_from_api
    participant WeatherAPI
    participant TrafficAPI

    Main->>plan_week: plan_week()の呼び出し
    activate plan_week

    plan_week->>data_integrator: data_integrator()の呼び出し
    activate data_integrator

    data_integrator->>fetch_data_from_api: fetch_data_from_api("/api/weather")
    activate fetch_data_from_api
    fetch_data_from_api->>WeatherAPI: GET /api/weather
    WeatherAPI-->>fetch_data_from_api: 天気データ
    fetch_data_from_api-->>data_integrator: 天気データ
    deactivate fetch_data_from_api

    data_integrator->>fetch_data_from_api: fetch_data_from_api("/api/traffic")
    activate fetch_data_from_api
    fetch_data_from_api->>TrafficAPI: GET /api/traffic
    TrafficAPI-->>fetch_data_from_api: 交通データ
    fetch_data_from_api-->>data_integrator: 交通データ
    deactivate fetch_data_from_api

    data_integrator-->>plan_week: 統合されたデータ
    deactivate data_integrator

    loop 各日のデータに対して
        plan_week->>plan_week: 日付の確認
        plan_week->>plan_week: 天気と交通の状況に基づいて計画を生成
    end

    plan_week-->>Main: 生成された計画
    deactivate plan_week

    Main->>Main: 計画の出力
```

```python
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サーバーを起動します：

```bash
python api_server.py
```

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

```bash
python data_integrator.py
```

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

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

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

## パフォーマンスと拡張性

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/24560849-a011-9532-1fdc-3f8fabb7da98.png)

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

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

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

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

### システムの拡張例

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

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

```python
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
```

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

```python
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
        }
```

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

```python
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
```

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

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

## まとめ

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2648069/349ef72a-dfe7-94d3-b924-02bc5e36d8dd.png)

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

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

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

Happy coding and smart living!