Python | 自動化：タイムアウト処理

Python
スポンサーリンク

概要（タイムアウト処理は「いつまでも待たない」ための保険）

タイムアウト処理は、

「相手（外部サービス・コマンド・ファイル処理など）がいつまでたっても返事してこないときに、
一定時間で“もう待つのをやめる”仕組み」

です。

自動化では、タイムアウトを入れていないと、

  • 外部APIが固まって、バッチが何時間もフリーズ
  • ネットワーク不調で無限に待ち続ける
  • 1件の失敗のせいで他の処理まで全部止まる

みたいな事故が普通に起こります。

だから「いつまで待つか」「待ち切れなかったらどうするか」を、
コードの中にきちんと書いておくことがすごく大事になります。

タイムアウト処理の考え方（まずは頭の整理から）

タイムアウト処理は、厳密に言うと「3つの要素」の組み合わせです。

ひとつ目は「何に対してタイムアウトをかけるのか」。
HTTP リクエスト、サブプロセス、非同期タスク、ファイル処理など。

ふたつ目は「どれくらいの時間で諦めるか」。
1秒？ 10秒？ 1分？ これは業務やサービスの要件で決まります。

みっつ目は「タイムアウトしたときにどう振る舞うか」。
リトライするのか、その件だけスキップするのか、バッチ全体を止めるのか。

この3つを「なんとなく」ではなく、最初に言葉で決めてからコードに落とすと、設計が一気に安定します。

HTTP リクエストのタイムアウト（requests の基本）

なぜ「timeout=なし」は絶対に避けるべきか

requests でよく見かける悪い例が、タイムアウト指定なしのコードです。

import requests

resp = requests.get("https://example.com/api/data")
Python

これだと、サーバーが固まったり、ネットワークがおかしくなったりしたとき、
この行で「無限に待ち続ける」可能性があります。

自動化スクリプトやバッチ処理では、これは致命的です。
必ず timeout を指定する癖をつけてください。

timeout の書き方と、全体像の例

基本はこうです。

import requests

def fetch_with_timeout(url: str, timeout_sec: float = 5.0):
    resp = requests.get(url, timeout=timeout_sec)
    resp.raise_for_status()
    return resp.json()
Python

timeout は「秒」です。小数もOKです。
この指定は「接続確立＋レスポンス受信」の両方の合計時間のように扱われます（厳密には connect/read 分けもできますが、初心者のうちは一つの数字でOKです）。

実務的には、「何回かリトライしたうえで諦める」形によくします。

import time
import requests

def fetch_with_retry(url: str, timeout_sec: float = 5.0, retries: int = 3, delay_sec: float = 2.0):
    for i in range(1, retries + 1):
        try:
            resp = requests.get(url, timeout=timeout_sec)
            resp.raise_for_status()
            return resp.json()
        except (requests.Timeout, requests.ConnectionError) as e:
            print(f"[{i}/{retries}] タイムアウトまたは接続エラー: {e}")
            if i == retries:
                raise
            time.sleep(delay_sec)
Python

ここで大事なのは、

  • 「1回の待ち時間（timeout_sec）」と「何回試すか（retries）」を分けて考えること
  • タイムアウトは普通の Exception と同じように try/except で扱えること
  • 最後の試行まで失敗したら、ちゃんと例外を投げて「失敗した」ことを上位に伝えること

です。

サブプロセス（外部コマンド）のタイムアウト（subprocess.run）

外部コマンドも「固まる前提」で考える

Python から外部プログラムやコマンドを呼ぶときも、タイムアウトは超重要です。
subprocess.run には timeout 引数が用意されています。

import subprocess

def run_with_timeout(cmd: list[str], timeout_sec: float = 10.0):
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout_sec
        )
        return result.stdout
    except subprocess.TimeoutExpired:
        print(f"コマンドが {timeout_sec} 秒以内に終わりませんでした: {cmd}")
        raise
Python

timeout 秒を過ぎてもコマンドが終了しない場合、TimeoutExpired 例外が出ます。
これをキャッチして、ログを残したり、再試行したり、諦めて次の処理に進んだりするわけです。

タイムアウト後にプロセスをどう扱うか

TimeoutExpired が出た時、subprocess.run は子プロセスを自動で kill します（引数次第ですが）。
もし自分で Popen を使って低レベルに制御しているなら、timeout したタイミングで kill や terminate を明示的に呼ぶ必要があります。

自動化バッチでは、「タイムアウトで止まらないようにする」だけでなく、
「タイムアウトした子プロセスをきちんと始末しておく」ことも設計の一部です。

asyncio のタイムアウト（asyncio.wait_for）

非同期タスクに対して「時間制限をつける」

asyncio の世界では、コルーチンやタスクに対して
「この処理は最大何秒まで待つ」という制限を asyncio.wait_for でつけられます。

import asyncio

async def long_task():
    await asyncio.sleep(5)
    return "done"

async def main():
    try:
        result = await asyncio.wait_for(long_task(), timeout=2.0)
        print("結果:", result)
    except asyncio.TimeoutError:
        print("タイムアウトしました")

asyncio.run(main())
Python

この例だと、long_task は5秒かかりますが、wait_for の timeout が 2 秒なので、
2秒で asyncio.TimeoutError が発生します。

ポイントは、

  • wait_for で包むことで、「コルーチンそのもの」に時間制限をかけられる
  • タイムアウト時にはコルーチンにキャンセルが飛ぶ（asyncio.CancelledError）
  • 呼び出し側は asyncio.TimeoutError を捕まえて、どうするか決める

という流れです。

タイムアウトしても「他のタスクは生かしたまま」にする設計

複数の非同期タスクを並列に走らせているとき、
一部だけが遅すぎる場合に「そのタスクだけ諦めて他は続行」ということもあります。

例えば：

async def safe_task(task_coro, timeout_sec):
    try:
        return await asyncio.wait_for(task_coro, timeout=timeout_sec)
    except asyncio.TimeoutError:
        print("このタスクはタイムアウトしたので諦めます")
        return None
Python

みたいなラッパーを作っておき、
gather に渡すときにこの safe_task 経由にしてやる、というパターンもあります。

「この API は1秒以内に返ってこなかったらもう古いので、諦めて None にする」
みたいなビジネスルールを、タイムアウト処理として実装するイメージです。

自前タイマーを使ったタイムアウト風処理（例として）

単純な while ループ＋時間チェック

ライブラリが timeout 引数を持っていないケースでは、
自前で「何秒経ったか」を測って、ループを抜けることで「タイムアウト風」の制御を作れます。

import time

def wait_until_condition(condition_func, timeout_sec=5.0, interval_sec=0.5):
    start = time.time()
    while True:
        if condition_func():
            return True
        if time.time() - start > timeout_sec:
            return False
        time.sleep(interval_sec)
Python

例えば、「あるファイルが出現するのを最長10秒待つ」などに使えます。

from pathlib import Path

def wait_file(path: Path, timeout_sec=10):
    def exists():
        return path.exists()
    return wait_until_condition(exists, timeout_sec, 0.5)
Python

これは「タイムアウト機能が組み込まれていない処理」に対して、
時間制限を後付けするときの基本的なアイデアです。

設計の重要ポイント（タイムアウトを「適当に決めない」）

どれくらい待つのが「正しい」のか

タイムアウト時間は、なんとなく 3 秒、5 秒、と決めると危険です。
本当は、ビジネスやシステムの要件から逆算されるべき数字です。

たとえば、

  • ユーザー向け画面なら「3秒以内に応答がないとイライラ」
  • バッチ処理なら「1つの API は最大30秒まで待ってよい」
  • 社内システム連携なら「1分までなら許容、それ以上は諦めてログに残す」

など、現場やサービスごとに「許容できる遅さ」が違います。

まず人間の都合・業務の都合を言葉にしてから、その時間をコードに埋めるのが理想です。

タイムアウト時の方針を最初に決めておく

タイムアウトの「時間」よりも、実は「タイムアウトしたときどうするか」が重要です。

例えば HTTP タイムアウトなら、

  • そのリクエストだけリトライしてみる（最大3回）
  • それでもダメなら「その1件だけスキップして、残りは続行」
  • 一定件数以上失敗したら、バッチ全体をエラー終了にする

こういうポリシーを先に決めておくと、例外処理のコードが書きやすくなります。

設計の順番としては、

  1. どの操作にタイムアウトをかけるか
  2. 何秒でタイムアウトさせるか
  3. タイムアウトしたときにどう振る舞うか

この3つを紙やコメントに書き出してから、コードを書き始めるとブレません。

ログに「タイムアウトが起きたこと」を必ず残す

自動化の現場では、「いつ・どの処理で・どれくらいの頻度でタイムアウトが起きているか」が運用上めちゃくちゃ重要です。

print でもいいので最初は必ずログを出しましょう。
本気運用なら logging モジュールでファイルに出すのがおすすめです。

import logging

logging.warning(f"APIタイムアウト: url={url}, timeout={timeout_sec}")
Python

こうしておけば、後から「timeout 3 秒は短すぎないか」「最近タイムアウトが増えていないか」などを振り返れます。

まとめ（タイムアウトは「止まらない自動化」の必須パーツ）

Python のタイムアウト処理を、自動化の文脈でまとめるとこうなります。

requests では必ず timeout= を指定し、必要ならリトライと組み合わせる。
subprocess.run では timeout= で外部コマンドの暴走を防ぎ、TimeoutExpired を捕まえてログや後処理を行う。
asyncio では asyncio.wait_for を使ってコルーチンに時間制限をかけ、TimeoutError を軸に制御する。
タイムアウト時間と、タイムアウト時の方針（リトライ・スキップ・全体停止）を最初に設計してからコードに落とす。
「タイムアウトが起きたこと」「どこでどれくらい起きたか」をログに残し、後から見直せるようにする。

タイムアウト処理は、「うまくいっているとき」には存在を忘れられる地味な機能です。
でも、自動化が本番に近づくほど、「何かがおかしくなったときに、止まらず・壊れず・ちゃんと知らせてくれる」ための最後の砦になります。

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