基礎から学ぶPython入門 90日コース | Web・API － Day 59：APIプロジェクト設計

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

Day 59：APIプロジェクト設計 ― 「ひとつのAPIツール」を最後まで組み立ててみる

Day 59では、これまで学んできた Web・API の知識をまとめて、 ひとつの「APIプロジェクト」を設計する視点 を身につけていきます。

キーワードはこの6つです。

  • 入力
  • API呼び出し
  • データ変換
  • 保存
  • エラー処理
  • ログ

今日は「部品ごとの書き方」だけでなく、 それらをどうつなげて「ひとつのツール」にするか を意識しながら読んでみてください。

全体像を先に描く：APIツールの基本フロー

APIプロジェクトの典型的な流れ

まずは、APIツールの全体像をざっくり言葉で並べます。

  1. 入力 ユーザーからキーワードやIDを受け取る／設定を読み込む
  2. API呼び出し 外部APIにリクエストを送り、レスポンスを受け取る
  3. データ変換 レスポンスJSONを「使いやすい形」に整える
  4. 保存 CSV・JSON・SQLiteなどに保存して、後から再利用できるようにする
  5. エラー処理 通信エラーやHTTPエラー、想定外のレスポンスに備える
  6. ログ 何が起きたかを記録し、トラブル時に原因を追えるようにする

この「6つの箱」を意識しながら、 ひとつのAPIプロジェクトを組み立てていきます。

入力：ツールに「どう指示を出すか」を決める

コマンドライン引数で検索キーワードを受け取る

例として、「検索キーワードを指定してAPIを叩くツール」 を作ることにします。

まずは、コマンドライン引数からキーワードを受け取る関数です。

# day59_input.py
import sys


def get_search_keyword() -> str:
    """
    コマンドライン引数から検索キーワードを取得する関数です。
    例: python main.py python 入門
    """

    # sys.argv[0] はスクリプト名、それ以降が引数です。
    args = sys.argv[1:]

    if not args:
        # 引数がない場合の扱いを決めておきます。
        print("検索キーワードが指定されていません。デフォルトのキーワードを使用します。")
        return "python"

    # 複数単語がある場合はスペースで結合してひとつのキーワードにします。
    keyword = " ".join(args)
    print(f"検索キーワード: {keyword}")
    return keyword


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

ポイントの深掘り：

  • 入力がないときにどうするか（デフォルト値にするか、エラーにするか）は、ツールの性格を決める大事な設計です。
  • 「入力をひとつの関数にまとめる」ことで、後から「設定ファイルも読む」「環境変数も見る」といった拡張がしやすくなります。

API呼び出し：外部サービスとの「窓口」をまとめる

Requestsを使ったシンプルなAPIクライアント

次に、外部APIを呼び出す部分です。 ここでは、requests を使って「検索APIっぽいエンドポイント」を叩くクラスを作ります。

# day59_api_client.py
import requests


class SearchApiClient:
    """
    検索APIを呼び出すためのクライアントクラスです。
    API呼び出しのロジックをひとつの場所にまとめておくことで、
    プロジェクト全体の見通しが良くなります。
    """

    def __init__(self, base_url: str):
        # ベースとなるAPIのURLを保持します。
        self.base_url = base_url

    def search(self, keyword: str) -> dict:
        """
        検索APIを呼び出して結果を返すメソッドです。
        実際には GET /search?q=keyword のようなイメージです。
        """

        params = {"q": keyword}
        print(f"[API] 検索APIを呼び出します: {self.base_url}, params={params}")

        # timeout を付けて、いつまでも待ち続けないようにします。
        response = requests.get(self.base_url, params=params, timeout=10)

        # HTTPステータスコードをチェックします。
        if response.status_code != 200:
            # ここでは簡易的に例外を投げますが、後でエラー処理を拡張します。
            raise RuntimeError(f"APIエラー: status={response.status_code}")

        # JSONレスポンスを辞書として返します。
        return response.json()
Python

ポイントの深掘り：

  • API呼び出しをクラスにまとめることで、「認証ヘッダーを追加する」「別のエンドポイントを増やす」といった拡張がしやすくなります。
  • timeout を付けるのは、ネットワーク系のツールではほぼ必須の習慣です。
  • 実務では response.raise_for_status() を使うことも多く、HTTPエラーを自動的に例外に変換できます。

データ変換：レスポンスJSONを「使いやすい形」に整える

必要な項目だけを抜き出す変換関数

APIレスポンスは、情報が多すぎてそのままだと扱いづらいことがよくあります。 そこで、必要な項目だけを抜き出して整形する層 を作ります。

# day59_transformer.py


def transform_search_results(raw_response: dict) -> list[dict]:
    """
    検索APIの生のレスポンスから、
    必要な項目だけを抜き出して整形する関数です。

    想定する変換後の形:
    - title: 結果のタイトル
    - url: リンク先URL
    - snippet: 簡単な説明文
    """

    print("[TRANSFORM] 検索結果を整形します。")

    # レスポンスの中から、結果リストを取り出します。
    items = raw_response.get("items", [])
    results: list[dict] = []

    for item in items:
        # .get() を使うことで、キーがなくても安全に値を取得できます。
        title = item.get("title", "")
        url = item.get("link", "")
        snippet = item.get("snippet", "")

        results.append(
            {
                "title": title,
                "url": url,
                "snippet": snippet,
            }
        )

    print(f"[TRANSFORM] {len(results)} 件の結果を整形しました。")
    return results
Python

ポイントの深掘り：

  • 「生のレスポンス」と「整形済みデータ」を分けて考えることで、コードの役割がはっきりします。
  • 変換ロジックをひとつの関数に閉じ込めておくと、API仕様が変わったときもここだけ直せば済む、という状態を作れます。
  • .get() にデフォルト値を指定しておくと、「一部の項目が欠けているレスポンス」にも強くなります。

保存：集めたデータを「残せる形」にする

CSV保存のユーティリティ関数

Day 56で扱ったように、保存形式はいくつか選べますが、 ここでは例として CSV保存 を使います。

# day59_storage.py
import csv
from pathlib import Path


def save_results_to_csv(results: list[dict], filename: str) -> Path:
    """
    検索結果をCSVファイルに保存する関数です。
    - results: title, url, snippet を持つ辞書のリスト
    - filename: 保存するファイル名
    """

    output_path = Path(filename)
    fieldnames = ["title", "url", "snippet"]

    print(f"[SAVE] 結果をCSVに保存します: {output_path}")

    # newline="" を指定して、余計な空行が入るのを防ぎます。
    with output_path.open(mode="w", encoding="utf-8", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()

        for item in results:
            writer.writerow(item)

    print(f"[SAVE] 保存完了: {output_path}")
    return output_path
Python

ポイントの深掘り：

  • 保存処理をひとつの関数にしておくと、「CSV → JSON → SQLite」と切り替えたいときに差し替えやすくなります。
  • Path を使うと、ファイルパスの扱いが直感的になり、OS差異も吸収しやすくなります。
  • 「保存先のパスを返す」ようにしておくと、後でログに書いたり、別の処理に渡したりしやすくなります。

エラー処理：トラブルが起きたときの「振る舞い」を決める

例外をキャッチして、ユーザーに分かる形で伝える

APIプロジェクトでは、 エラーが起きない前提で書かれたコードはすぐ壊れます。

そこで、メインの流れの中に try〜except を組み込んでおきます。

# day59_main_with_error.py
from day59_input import get_search_keyword
from day59_api_client import SearchApiClient
from day59_transformer import transform_search_results
from day59_storage import save_results_to_csv


def run_search_tool() -> None:
    """
    検索APIプロジェクトのメイン処理です。
    入力 → API呼び出し → データ変換 → 保存 の流れを、
    エラー処理を含めてひとつにまとめています。
    """

    # 1. 入力
    keyword = get_search_keyword()

    # 2. APIクライアントの準備
    base_url = "https://example.com/api/search"  # 実際のAPI URLに置き換えます。
    client = SearchApiClient(base_url=base_url)

    try:
        # 3. API呼び出し
        raw_response = client.search(keyword)

        # 4. データ変換
        results = transform_search_results(raw_response)

        if not results:
            print("検索結果が0件でした。保存は行いません。")
            return

        # 5. 保存
        save_results_to_csv(results, "search_results.csv")

    except Exception as e:
        # 6. エラー処理
        print("[ERROR] 処理中にエラーが発生しました。")
        print(f"詳細: {e}")


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

ポイントの深掘り：

  • 「どこで例外が起きても、最後にまとめてキャッチする」形にしておくと、ツールが途中でクラッシュしにくくなります。
  • ユーザーに見せるメッセージと、開発者が見る詳細（スタックトレースなど）を分けるのが理想的です。
  • 実務では、ここに「再試行」「代替処理」「通知」などを追加していきます。

ログ：あとから「何が起きたか」を追えるようにする

logging モジュールでコンソール＋ファイルに記録する

最後に、「ログ」です。 ログは、「あとから振り返るための記録」 です。

# day59_logging_setup.py
import logging
from pathlib import Path


def setup_logger(log_file: str = "search_tool.log") -> logging.Logger:
    """
    ロガーを設定して返す関数です。
    コンソールとファイルの両方にログを出力するように設定します。
    """

    logger = logging.getLogger("search_tool")
    logger.setLevel(logging.INFO)

    # すでにハンドラが設定されている場合は何もしません（重複防止）。
    if logger.handlers:
        return logger

    # コンソール出力用ハンドラ
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)

    # ファイル出力用ハンドラ
    file_handler = logging.FileHandler(Path(log_file), encoding="utf-8")
    file_handler.setLevel(logging.INFO)

    # ログのフォーマットを設定します。
    formatter = logging.Formatter(
        "[%(asctime)s] %(levelname)s %(name)s - %(message)s"
    )
    console_handler.setFormatter(formatter)
    file_handler.setFormatter(formatter)

    logger.addHandler(console_handler)
    logger.addHandler(file_handler)

    return logger
Python

このロガーをメイン処理で使ってみます。

# day59_main_with_logging.py
from day59_input import get_search_keyword
from day59_api_client import SearchApiClient
from day59_transformer import transform_search_results
from day59_storage import save_results_to_csv
from day59_logging_setup import setup_logger


def run_search_tool() -> None:
    """
    ログ付きの検索APIプロジェクトメイン処理です。
    """

    logger = setup_logger()

    # 1. 入力
    keyword = get_search_keyword()
    logger.info(f"検索ツール開始: keyword='{keyword}'")

    # 2. APIクライアントの準備
    base_url = "https://example.com/api/search"
    client = SearchApiClient(base_url=base_url)

    try:
        # 3. API呼び出し
        logger.info("API呼び出しを開始します。")
        raw_response = client.search(keyword)
        logger.info("API呼び出しが成功しました。")

        # 4. データ変換
        results = transform_search_results(raw_response)
        logger.info(f"整形された結果件数: {len(results)}")

        if not results:
            logger.warning("検索結果が0件でした。保存は行いません。")
            return

        # 5. 保存
        output_path = save_results_to_csv(results, "search_results.csv")
        logger.info(f"結果を保存しました: {output_path}")

    except Exception:
        # 6. エラー処理＋ログ
        logger.error("処理中にエラーが発生しました。", exc_info=True)
        print("[ERROR] 処理中にエラーが発生しました。詳細はログファイルを確認してください。")


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

ポイントの深掘り：

  • logger.info() で「正常な流れ」を記録しておくと、後から「どこまで進んでいたか」が分かります。
  • logger.error(..., exc_info=True) で例外の詳細もログに残せるので、トラブルシューティングがしやすくなります。
  • コンソールとファイルの両方に出すことで、「開発中の確認」と「運用時の記録」を両立できます。

Day 59ミニテンプレート：APIプロジェクト設計の「ひとまとめコード」

最後に、Day 59の内容をコンパクトにまとめた APIプロジェクト設計テンプレート を載せておきます。

# day59_api_project_template.py
import sys
import requests
import csv
import logging
from pathlib import Path


def get_search_keyword() -> str:
    args = sys.argv[1:]
    if not args:
        print("検索キーワードが指定されていません。デフォルト 'python' を使用します。")
        return "python"
    return " ".join(args)


class SearchApiClient:
    def __init__(self, base_url: str):
        self.base_url = base_url

    def search(self, keyword: str) -> dict:
        params = {"q": keyword}
        response = requests.get(self.base_url, params=params, timeout=10)
        # HTTPエラーなら例外を投げます。
        response.raise_for_status()
        return response.json()


def transform_search_results(raw_response: dict) -> list[dict]:
    items = raw_response.get("items", [])
    results: list[dict] = []
    for item in items:
        results.append(
            {
                "title": item.get("title", ""),
                "url": item.get("link", ""),
                "snippet": item.get("snippet", ""),
            }
        )
    return results


def save_results_to_csv(results: list[dict], filename: str) -> Path:
    path = Path(filename)
    fieldnames = ["title", "url", "snippet"]
    with path.open(mode="w", encoding="utf-8", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for item in results:
            writer.writerow(item)
    return path


def setup_logger(log_file: str = "search_tool.log") -> logging.Logger:
    logger = logging.getLogger("search_tool")
    logger.setLevel(logging.INFO)
    if logger.handlers:
        return logger

    console = logging.StreamHandler()
    file_handler = logging.FileHandler(Path(log_file), encoding="utf-8")

    formatter = logging.Formatter("[%(asctime)s] %(levelname)s %(name)s - %(message)s")
    console.setFormatter(formatter)
    file_handler.setFormatter(formatter)

    logger.addHandler(console)
    logger.addHandler(file_handler)
    return logger


def run_search_tool() -> None:
    logger = setup_logger()

    keyword = get_search_keyword()
    logger.info(f"検索ツール開始: keyword='{keyword}'")

    client = SearchApiClient(base_url="https://example.com/api/search")

    try:
        logger.info("API呼び出し開始")
        raw = client.search(keyword)
        logger.info("API呼び出し成功")

        results = transform_search_results(raw)
        logger.info(f"整形結果件数: {len(results)}")

        if not results:
            logger.warning("結果0件のため保存しません。")
            return

        path = save_results_to_csv(results, "search_results.csv")
        logger.info(f"結果保存: {path}")

    except Exception:
        logger.error("処理中にエラー発生", exc_info=True)
        print("[ERROR] エラーが発生しました。詳細はログを確認してください。")


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

Day 59のまとめ

Day 59では、APIプロジェクト設計の基礎として、

  • 入力 → API呼び出し → データ変換 → 保存 → エラー処理 → ログ という「ひとつの流れ」を意識してツールを組み立てること
  • 入力設計がツールの使い勝手を決めること
  • API呼び出しをクライアントクラスにまとめることで、見通しと再利用性が上がること
  • データ変換層を作ることで、レスポンスJSONを「使いやすい形」にできること
  • 保存処理をユーティリティ関数にしておくと、形式の切り替えがしやすいこと
  • エラー処理とログを組み込むことで、「壊れにくく、原因を追いやすい」ツールになること

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

ここまで来ると、「ちょっとしたAPIツール」を思いついたときに、 どんな部品をどう並べればいいか が、かなりクリアに見えてくるはずです。

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