基礎から学ぶPython入門 90日コース | Web・API － Day 58：複数API連携

Python 90日で身につけるPython
スポンサーリンク
スポンサーリンク

Day 58：複数API連携 ― 「バラバラの情報」をひとつのデータにまとめる

Day 58では、複数のAPIを組み合わせて使う ことをテーマに、 実際に「複数APIデータ統合ツール」を作っていきます。

やりたいことはシンプルです。

  • 複数のAPIからデータを取得する
  • それぞれの結果を「ひとつのまとまったデータ」に統合する
  • ツールとして、毎回同じ流れで使える形にする

Web・APIの世界では、 「1つの画面を作るために、裏側で複数のAPIを呼んでいる」 というのはよくある話です。

今日は、その裏側の流れをPythonで再現してみましょう。

複数API連携のイメージを言葉で整理する

ひとつの「統合データ」を作るシナリオ

まずは、どんなAPIを組み合わせるか、 イメージしやすいシナリオを決めます。

例として、次のような「ユーザー情報統合」を考えてみます。

  1. ユーザー基本情報API
    • name（名前）
    • email（メールアドレス）
  2. ユーザー活動情報API
    • last_login（最終ログイン日時）
    • login_count（ログイン回数）
  3. ユーザー購入情報API
    • last_order_id（最後の注文ID）
    • total_spent（累計購入金額）

これらをまとめて、

{
    "name": "Alice",
    "email": "alice@example.com",
    "last_login": "2024-01-01T12:34:56",
    "login_count": 42,
    "last_order_id": "ORD-12345",
    "total_spent": 12345.67,
}
Python

のような 統合されたユーザーデータ を作るイメージです。

サンプルAPIレスポンスを用意する（モックデータ）

実際のAPIの代わりに「モック関数」を使う

ここでは、実際の外部APIではなく、 「APIっぽい関数」 を使って説明していきます。

# day58_mock_apis.py

def fetch_user_profile(user_id: int) -> dict:
    """
    ユーザー基本情報APIのモックです。
    実際にはHTTPリクエストを送るところですが、ここでは固定データを返します。
    """
    print(f"[API] ユーザー基本情報を取得します (user_id={user_id})")
    return {
        "user_id": user_id,
        "name": "Alice",
        "email": "alice@example.com",
    }


def fetch_user_activity(user_id: int) -> dict:
    """
    ユーザー活動情報APIのモックです。
    """
    print(f"[API] ユーザー活動情報を取得します (user_id={user_id})")
    return {
        "user_id": user_id,
        "last_login": "2024-01-01T12:34:56",
        "login_count": 42,
    }


def fetch_user_orders(user_id: int) -> dict:
    """
    ユーザー購入情報APIのモックです。
    """
    print(f"[API] ユーザー購入情報を取得します (user_id={user_id})")
    return {
        "user_id": user_id,
        "last_order_id": "ORD-12345",
        "total_spent": 12345.67,
    }
Python

この3つの関数を「APIの代わり」として使いながら、 複数API連携の流れを組み立てていきます。

ステップ1：同期的に複数APIを呼び出して統合する

まずは「順番に呼んで、まとめる」基本形

最初のステップでは、 同期処理で順番にAPIを呼び出して統合する 流れを作ります。

# day58_sync_integration.py
from day58_mock_apis import (
    fetch_user_profile,
    fetch_user_activity,
    fetch_user_orders,
)


def integrate_user_data_sync(user_id: int) -> dict:
    """
    複数APIからユーザーデータを同期的に取得し、
    ひとつの辞書に統合する関数です。
    """

    print(f"[SYNC] ユーザーデータ統合を開始します (user_id={user_id})")

    # 1. ユーザー基本情報を取得
    profile = fetch_user_profile(user_id)

    # 2. ユーザー活動情報を取得
    activity = fetch_user_activity(user_id)

    # 3. ユーザー購入情報を取得
    orders = fetch_user_orders(user_id)

    # 4. 統合データを作成
    integrated = {
        "user_id": user_id,
        "name": profile["name"],
        "email": profile["email"],
        "last_login": activity["last_login"],
        "login_count": activity["login_count"],
        "last_order_id": orders["last_order_id"],
        "total_spent": orders["total_spent"],
    }

    print("[SYNC] ユーザーデータ統合が完了しました。")
    return integrated


def main():
    user_id = 1
    integrated_data = integrate_user_data_sync(user_id)

    print("\n=== 統合されたユーザーデータ ===")
    for key, value in integrated_data.items():
        print(f"{key}: {value}")


if __name__ == "__main__":
    main()
Python

ポイント：

  • 3つのAPIを 順番に 呼び出しています。
  • それぞれの結果から必要な項目を取り出して、ひとつの辞書にまとめています。
  • 「統合ツール」として、integrate_user_data_sync() が入口になっています。

この形は、まず「複数API連携の基本」を理解するにはとても分かりやすいです。

ステップ2：非同期版の複数API連携を考える

「待ち時間があるAPI」をまとめて呼びたいとき

Day 57で学んだように、 非同期処理は 「待っている時間をうまく使う」 ための仕組みです。

複数API連携では、

  • 各APIのレスポンスを待っている時間
  • ネットワークの待ち時間

が積み重なることが多いので、 非同期処理と相性が良い場面がたくさんあります。

ここでは、asyncio を使って 「非同期版のモックAPI」と「非同期統合関数」を作ってみます。

非同期モックAPIを用意する

async と await を使ったAPI風関数

# day58_async_mock_apis.py
import asyncio


async def fetch_user_profile_async(user_id: int) -> dict:
    """
    非同期版のユーザー基本情報APIモックです。
    asyncio.sleep() で「ネットワーク待ち」を模しています。
    """
    print(f"[ASYNC API] ユーザー基本情報を取得します (user_id={user_id})")
    await asyncio.sleep(1)  # 1秒待つ（ネットワーク待ちのイメージ）
    return {
        "user_id": user_id,
        "name": "Alice",
        "email": "alice@example.com",
    }


async def fetch_user_activity_async(user_id: int) -> dict:
    """
    非同期版のユーザー活動情報APIモックです。
    """
    print(f"[ASYNC API] ユーザー活動情報を取得します (user_id={user_id})")
    await asyncio.sleep(1.5)  # 1.5秒待つ
    return {
        "user_id": user_id,
        "last_login": "2024-01-01T12:34:56",
        "login_count": 42,
    }


async def fetch_user_orders_async(user_id: int) -> dict:
    """
    非同期版のユーザー購入情報APIモックです。
    """
    print(f"[ASYNC API] ユーザー購入情報を取得します (user_id={user_id})")
    await asyncio.sleep(2)  # 2秒待つ
    return {
        "user_id": user_id,
        "last_order_id": "ORD-12345",
        "total_spent": 12345.67,
    }
Python

ここでは、

  • await asyncio.sleep() を使って「ネットワーク待ち」を模しています。
  • 実際のHTTPクライアント（aiohttp や httpx の非同期版など）を使うときも、 同じように await でレスポンスを待つ形になります。

非同期で複数APIをまとめて呼び出して統合する

asyncio.gather() で「まとめて待つ」

非同期版の統合関数では、 3つのAPIを ほぼ同時に呼び出して、まとめて待つ 形にしてみます。

# day58_async_integration.py
import asyncio
from day58_async_mock_apis import (
    fetch_user_profile_async,
    fetch_user_activity_async,
    fetch_user_orders_async,
)


async def integrate_user_data_async(user_id: int) -> dict:
    """
    複数APIからユーザーデータを非同期的に取得し、
    ひとつの辞書に統合する関数です。
    """

    print(f"[ASYNC] ユーザーデータ統合を開始します (user_id={user_id})")

    # 3つのAPI呼び出しを「まとめて」スケジュールします。
    profile_task = asyncio.create_task(fetch_user_profile_async(user_id))
    activity_task = asyncio.create_task(fetch_user_activity_async(user_id))
    orders_task = asyncio.create_task(fetch_user_orders_async(user_id))

    # 3つのタスクがすべて完了するまで待ちます。
    # asyncio.gather() を使うと、複数の await をまとめて扱えます。
    profile, activity, orders = await asyncio.gather(
        profile_task, activity_task, orders_task
    )

    # 統合データを作成します。
    integrated = {
        "user_id": user_id,
        "name": profile["name"],
        "email": profile["email"],
        "last_login": activity["last_login"],
        "login_count": activity["login_count"],
        "last_order_id": orders["last_order_id"],
        "total_spent": orders["total_spent"],
    }

    print("[ASYNC] ユーザーデータ統合が完了しました。")
    return integrated


async def main_async():
    user_id = 1
    integrated_data = await integrate_user_data_async(user_id)

    print("\n=== 統合されたユーザーデータ（非同期版） ===")
    for key, value in integrated_data.items():
        print(f"{key}: {value}")


if __name__ == "__main__":
    asyncio.run(main_async())
Python

ポイント：

  • asyncio.create_task() で、3つのAPI呼び出しを「同時に走らせる」イメージでスケジュールしています。
  • asyncio.gather() で、3つのタスクが全部終わるまで待ち、結果をまとめて受け取っています。
  • 統合のロジック自体は同期版とほぼ同じで、「取得方法だけが非同期になっている」形です。

実際のAPI呼び出しがそれぞれ1〜2秒かかるとすると、

  • 同期版：1秒＋1.5秒＋2秒 ≒ 4.5秒
  • 非同期版：最大の2秒にほぼ近い時間で完了

という違いが出てきます（あくまでイメージですが）。

複数APIデータ統合ツールの「使いやすい形」を整える

統合ツールとしての入口関数を用意する

複数API連携をツールとして使うときは、

  • 統合処理を行う関数
  • 結果を保存したり表示したりする部分

を分けておくと、後から拡張しやすくなります。

# day58_integration_tool.py
import asyncio
from day58_async_integration import integrate_user_data_async
from day58_sync_integration import integrate_user_data_sync


def run_sync_tool(user_id: int) -> None:
    """
    同期版の複数APIデータ統合ツールを実行する関数です。
    """
    print("=== 同期版 複数APIデータ統合ツール ===")
    integrated = integrate_user_data_sync(user_id)

    print("\n--- 統合結果（同期版） ---")
    for key, value in integrated.items():
        print(f"{key}: {value}")


async def run_async_tool(user_id: int) -> None:
    """
    非同期版の複数APIデータ統合ツールを実行する関数です。
    """
    print("=== 非同期版 複数APIデータ統合ツール ===")
    integrated = await integrate_user_data_async(user_id)

    print("\n--- 統合結果（非同期版） ---")
    for key, value in integrated.items():
        print(f"{key}: {value}")


def main():
    user_id = 1

    # 同期版ツールを実行
    run_sync_tool(user_id)

    print("\n")

    # 非同期版ツールを実行
    asyncio.run(run_async_tool(user_id))


if __name__ == "__main__":
    main()
Python

この形にしておくと、

  • 同期版と非同期版の違いを簡単に比較できる
  • 将来的に「結果をCSV/JSON/SQLiteに保存する」処理を追加しやすい
  • ツールとして「ユーザーIDを変えて実行する」などの拡張もしやすい

というメリットがあります。

Day 58ミニテンプレート：複数APIデータ統合ツールの基本形

最後に、Day 58の内容をコンパクトにまとめた 「複数APIデータ統合ツール」テンプレート を載せておきます。

# day58_multi_api_integration_template.py
import asyncio


async def fetch_api1(user_id: int) -> dict:
    """API1のモック。"""
    await asyncio.sleep(1)
    return {"user_id": user_id, "name": "Alice"}


async def fetch_api2(user_id: int) -> dict:
    """API2のモック。"""
    await asyncio.sleep(1.5)
    return {"user_id": user_id, "last_login": "2024-01-01T12:34:56"}


async def fetch_api3(user_id: int) -> dict:
    """API3のモック。"""
    await asyncio.sleep(2)
    return {"user_id": user_id, "total_spent": 12345.67}


async def integrate_multi_api(user_id: int) -> dict:
    """
    複数APIからデータを取得して統合する基本テンプレートです。
    """

    # 3つのAPI呼び出しをまとめてスケジュール
    t1 = asyncio.create_task(fetch_api1(user_id))
    t2 = asyncio.create_task(fetch_api2(user_id))
    t3 = asyncio.create_task(fetch_api3(user_id))

    # 3つの結果をまとめて待つ
    r1, r2, r3 = await asyncio.gather(t1, t2, t3)

    # 統合データを作成
    integrated = {
        "user_id": user_id,
        "name": r1["name"],
        "last_login": r2["last_login"],
        "total_spent": r3["total_spent"],
    }

    return integrated


async def main_async():
    user_id = 1
    data = await integrate_multi_api(user_id)

    print("=== 複数API統合結果 ===")
    for key, value in data.items():
        print(f"{key}: {value}")


if __name__ == "__main__":
    asyncio.run(main_async())
Python

このテンプレートは、

  • 複数API呼び出し
  • 非同期処理 (async / await)
  • 統合データの作成

という「複数API連携の基本セット」を、 コンパクトに体験できる形になっています。

Day 58のまとめ

Day 58では、複数API連携の基礎として、

  • 複数のAPIからバラバラに返ってくる情報を、ひとつの「統合データ」にまとめる考え方
  • 同期版として、順番にAPIを呼び出して統合する基本形
  • 非同期版として、async / await と asyncio.gather() を使って、複数APIをまとめて呼び出す形
  • 「複数APIデータ統合ツール」として、統合処理をひとつの関数にまとめて使いやすくする形

を、コード例とともにステップバイステップで整理しました。

この土台があると、

  • ユーザー情報＋活動ログ＋購入履歴をまとめたダッシュボード
  • 複数の外部サービスからデータを集めて統合するレポートツール
  • マイクロサービス間の連携を行う「集約API」

など、より実践的な「複数API連携」の世界へ、 自然にステップアップしていけるようになります。

スポンサーリンク
タイトルとURLをコピーしました