Python | 自動化：API 定期取得

Python
スポンサーリンク

概要（API定期取得は「決まった時間にデータを取りに行くロボット」）

API定期取得は、
「毎朝9時にレポートAPIを叩く」「10分ごとにセンサーAPIから値を取る」
といった処理を Python で自動実行する仕組みです。

やること自体はシンプルです。

1回分の「API呼び出し＋保存」の処理を書いたPythonスクリプトを作る。
そのスクリプトを cron（Linux / Mac）やタスクスケジューラ（Windows）で定期実行する。

大事なのは「1回分の処理」をちゃんと設計することです。
エラー時の扱い・ログ・保存形式・再実行のしやすさを丁寧にしておくと、一気に“本番っぽく”なります。

1回分の処理を作る（API を叩いて保存する、最小のスクリプト）

ライブラリとフォルダ構成を決める

APIを叩く定番は requests です。

pip install requests

フォルダはこんなイメージがおすすめです。

  • project_root/
    • api_fetch.py（メインスクリプト）
    • logs/（ログ）
    • data/（取得データ）

絶対パスを扱いやすくするため、Path(__file__).parent を基準にします。

最小の「1回分 API 取得＋保存」コード

まずは「JSONを取得して、そのままファイルに保存」するだけのサンプル。

# api_fetch.py
import json
import logging
from datetime import datetime
from pathlib import Path

import requests

BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data"
LOG_FILE = BASE_DIR / "logs" / "api_fetch.log"

API_URL = "https://api.example.com/v1/data"  # ここはあなたのAPIに変える

def setup_logging():
    LOG_FILE.parent.mkdir(exist_ok=True)
    logging.basicConfig(
        filename=LOG_FILE,
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s"
    )

def fetch_from_api():
    logging.info(f"APIリクエスト開始: {API_URL}")
    resp = requests.get(API_URL, timeout=10)
    resp.raise_for_status()  # ステータスコードが4xx/5xxなら例外
    data = resp.json()
    logging.info("APIレスポンス取得成功")
    return data

def save_json(data):
    DATA_DIR.mkdir(exist_ok=True)
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    out_path = DATA_DIR / f"raw_{ts}.json"
    with out_path.open("w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    logging.info(f"JSON保存完了: {out_path}")

def run_job():
    data = fetch_from_api()
    save_json(data)

def main():
    logging.info("API定期取得バッチ開始")
    try:
        run_job()
        logging.info("API定期取得バッチ正常終了")
    except Exception as e:
        logging.exception(f"API定期取得バッチ異常終了: {e}")
        raise

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

ここで深掘りしたいポイントは4つあります。

1つ目は BASE_DIR = Path(__file__).resolve().parent です。
「スクリプトが置かれているフォルダ」を基準にしているので、どこから実行しても崩れません。

2つ目は logging の初期化です。
成功・失敗・何が起きたかをファイルに残しておくことで、夜中にエラーが起きても朝確認できます。

3つ目は resp.raise_for_status()。
HTTPステータスが 4xx/5xx のときに例外を投げ、logging.exception へつながります。
「エラーに気づけるかどうか」はここにかかっています。

4つ目は raw_YYYYMMDD_HHMMSS.json というファイル名です。
タイムスタンプを埋めることで「いつ取ったデータか」が一目でわかり、上書き事故も防げます。

認証付きAPI（APIキー・トークン）の扱い方

一番シンプルな「ヘッダーにAPIキーを乗せる」パターン

多くのAPIは、ヘッダーにキーやトークンを付けます。

API_URL = "https://api.example.com/v1/data"
API_KEY = "ここに本物のキーは直書きしない"

def fetch_from_api():
    headers = {"Authorization": f"Bearer {API_KEY}"}
    resp = requests.get(API_URL, headers=headers, timeout=10)
    resp.raise_for_status()
    return resp.json()
Python

ただし、本物のキーをソースに直書きするのは危険です。
Gitに上げた途端に漏洩、という事故は本当に多いです。

環境変数でAPIキーを渡す（安全寄り）

環境変数から読み込む形にしておくと、コードに秘密情報を残さずに済みます。

import os

API_KEY = os.environ.get("MY_API_KEY")

def fetch_from_api():
    if not API_KEY:
        raise RuntimeError("環境変数 MY_API_KEY が設定されていません")
    headers = {"Authorization": f"Bearer {API_KEY}"}
    resp = requests.get(API_URL, headers=headers, timeout=10)
    resp.raise_for_status()
    return resp.json()
Python

実行環境側（cron / タスクスケジューラ / サーバー設定）で
MY_API_KEY をセットしておく運用にすると安心です。

取得データの保存設計（そのままJSONか、整形してCSV / DBか）

JSONをそのまま残すメリット

まずは「生のレスポンスを、そのままJSONで残す」方針が無難です。

後から構造が変わったときに、「その時点のレスポンス」を見返せるからです。
API側の仕様変更や、不具合の調査にとても役立ちます。

解析しやすい形（CSV / DB）への変換

分析や集計をしたい場合は、JSONをパースしてから、
表形式（行×列）に落とし込みます。

簡単な例として「配列のJSON → CSV」を考えます。

import pandas as pd

def normalize_and_save(data):
    DATA_DIR.mkdir(exist_ok=True)
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")

    df = pd.json_normalize(data)  # dataがリスト / ネストありでも平坦化できる
    csv_path = DATA_DIR / f"normalized_{ts}.csv"
    df.to_csv(csv_path, index=False, encoding="utf-8-sig")
    logging.info(f"整形CSV保存完了: {csv_path}")
Python

そして run_job を

def run_job():
    data = fetch_from_api()
    save_json(data)         # 生のJSONを残す
    normalize_and_save(data)  # 解析しやすいCSVも作る
Python

のようにしておくと、「生データ」と「加工済みデータ」の両方を保存できます。

DBに永続化する（ログ・履歴として貯める）

履歴を長期保存したいなら、SQLite などのDBに入れるのも手です。
1テーブルに「取得結果＋取得日時」の形で追加していくと、
後から「いつどんな値だったか」を簡単に照会できます。

エラーとリトライ（ネットワーク前提なら必須の考え方）

一時的な失敗（ネットワーク切れ、タイムアウト）はリトライ

APIは必ずいつか失敗します。
ネットワークが落ちる、APIサーバーが不安定、タイムアウト…など。

一時的なエラーは「少し待ってもう一度やる」で回避できることが多いので、
簡単なリトライ処理を入れておきます。

import time
import requests

def fetch_from_api_with_retry(retries=3, delay=5):
    for i in range(1, retries + 1):
        try:
            logging.info(f"APIリクエスト試行 {i}/{retries}")
            resp = requests.get(API_URL, timeout=10)
            resp.raise_for_status()
            logging.info("APIレスポンス取得成功")
            return resp.json()
        except requests.RequestException as e:
            logging.warning(f"APIリクエスト失敗 ({i}/{retries}): {e}")
            if i == retries:
                logging.error("リトライ回数上限に達しました。失敗として終了します。")
                raise
            time.sleep(delay)
Python

こうしておけば、ネットワークが一瞬切れた程度では「バッチ全体が失敗」になりません。

過剰なリトライは「API側の迷惑」になることも

リトライは便利ですが、
無限ループにしたり、超短間隔で連打したりすると、API側の負荷になります。

APIの利用規約（rate limit）を確認し、
「最大回数」「間隔」を適切に決めることが大事です。

定期実行との接続（cron / タスクスケジューラ）

詳しい話はすでに話したので、ここでは「組み合わせ方」だけ。

Linux / Mac での例

api_fetch.py を毎時00分に実行するイメージ。

0 * * * * /home/you/project/venv/bin/python /home/you/project/api_fetch.py >> /home/you/project/logs/cron_stdout.log 2>&1

ポイントは、

  • 仮想環境の python をフルパスで指定する
  • スクリプトもフルパス
  • 出力をログにリダイレクトしておく

の3つです。

Windows のタスクスケジューラでの例

タスクスケジューラなら、

  • プログラム／スクリプト：C:\project\venv\Scripts\python.exe
  • 引数：C:\project\api_fetch.py
  • 開始（作業ディレクトリ）：C:\project
  • トリガー：毎時

という設定にしておきます。

スクリプト側では BASE_DIR = Path(__file__).resolve().parent を使っているので、
作業フォルダが変でも大丈夫ですが、「開始場所」を指定しておくと安定します。

設計で特に大事なポイントの深掘り

1. 「1回分の処理」をきれいに分ける

  • 環境設定＆ログ初期化（setup_logging）
  • API呼び出し（fetch_from_api）
  • データ整形（normalize_and_save）
  • 保存（ファイル or DB）（save_json など）
  • 例外ハンドリング付きの main（main）

この分解をしておくと、
あとからテスト・差し替え・拡張がやりやすくなります。
「全部を main にベタ書き」は後で必ず辛くなります。

2. 冪等性（再実行しても壊れない）を考える

もし1回失敗したときに「もう一度同じ日付のデータだけ再取得したい」となった場合、
同じスクリプトを再実行したときにどうなるかをイメージしてください。

例えば、ファイル名に日付だけを使っていると、
上書きされて過去のデータが消えます。それが意図ならOKですが、
「追記したい / 差分だけ入れたい」場合は保存のルールを変える必要があります。

  • タイムスタンプ付きファイル名にして全部残す
  • DBに入れる場合はユニークキーを設計し、重複時はUPDATEにする
  • 「日付単位で一旦削除してから新しいものを入れ直す」設計にする

こういう工夫をしておくと、「再実行が怖くない」バッチになります。

3. 「見えるログ」を最初から仕込む

定期実行は、うまく動いているときは存在を忘れがちですが、
こっそり失敗しているときが一番怖いです。

  • 開始時に「何のバッチを始めたか」
  • 対象日・対象パラメータ
  • 入出力ファイル・件数
  • 正常終了 or 異常終了のメッセージ

このあたりを INFO レベルで出しておくと、
あとからログをパッと眺めるだけで状況が分かります。

まとめ（「1回分の処理をきちんと作り、OSに“時間だけ”任せる）

API定期取得の本質は、難しいことではありません。

1回分の「API呼び出し＋保存」を

  • パスは絶対パスで
  • ログとエラー処理を入れて
  • 再実行しても壊れないように

丁寧に作る。

時間管理（毎朝、毎時など）は cron / タスクスケジューラに任せる。
Python側は「呼ばれたら1回きっちりやる」ことに集中する。

この形さえ掴めば、
「レポートAPIを毎朝取得」「センサーAPIを10分ごとに取得」「SNS APIから定期クローリング」など、
いくらでも応用が効くようになります。

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