1. はじめに
Pythonで処理がいつまでも終わらない場合や、外部のAPI・通信・重い計算などに時間制限を設けたいとき、「一定時間で自動終了する処理(タイムアウト処理)」が役立ちます。
本記事では、Python初心者〜中級者の方向けに、タイムアウト処理を簡単に実装する方法を解説します。標準ライブラリを使った基本の方法から、実務でも使える応用例、注意点までを丁寧に紹介します。
2. Pythonでタイムアウト処理を行う基本
2-1. タイムアウト処理とは?
タイムアウト処理とは、指定した時間内に処理が終わらなければ、自動的にその処理を打ち切る仕組みのことです。Pythonでは主に以下のような方法があります:
signal
モジュールを使う(UNIX/Linux向け)threading
+join()
を使うconcurrent.futures
を使う
ここでは、より汎用性の高いconcurrent.futures
による方法を中心に解説します。
2-2. concurrent.futuresでタイムアウト処理
まずは基本的な使い方を見てみましょう。
import concurrent.futures
import time
# タイムアウト対象の関数
def long_running_task():
time.sleep(10)
return "完了しました!"
# スレッドプールを使って非同期に実行
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(long_running_task)
try:
result = future.result(timeout=5) # 5秒でタイムアウト
print(result)
except concurrent.futures.TimeoutError:
print("タイムアウトしました。")
実行結果:
タイムアウトしました。
この例では、10秒かかる処理に対して5秒のタイムアウトを設定しているため、自動的に処理を中断して「タイムアウトしました。」と表示されます。
3. よくある使い方・応用例
3-1. 外部APIのタイムアウト処理
APIなど外部サービスとの通信では、応答が遅れるとアプリ全体が停止してしまう恐れがあります。以下のように、requests
ライブラリでもタイムアウト処理を組み込めます。
import requests
try:
response = requests.get("https://httpbin.org/delay/10", timeout=3)
print(response.status_code)
except requests.exceptions.Timeout:
print("リクエストがタイムアウトしました。")
実行結果:
リクエストがタイムアウトしました。
3-2. 任意の関数にタイムアウトを付与するユーティリティ
タイムアウト処理を汎用化したい場合、以下のような関数として使い回すことも可能です。
def run_with_timeout(func, timeout):
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(func)
try:
return future.result(timeout=timeout)
except concurrent.futures.TimeoutError:
return "処理がタイムアウトしました"
# 使用例
def example_task():
time.sleep(7)
return "成功!"
result = run_with_timeout(example_task, 5)
print(result)
実行結果:
処理がタイムアウトしました
4. 注意点・エラー対策
4-1. タイムアウト後の処理に注意
タイムアウトしてもスレッドやプロセスは完全に停止しないことがあります。そのため、次のような点に注意してください:
- タイムアウト後にリソースを自動解放しない
- スレッドがバックグラウンドで動き続ける可能性がある
特に無限ループや長時間の処理には、処理の中で定期的にチェックを入れると安全です。
4-2. signalモジュールはLinux限定
signal.signal
を使ったタイムアウト処理は便利ですが、Windowsでは動作しない点に注意が必要です。以下はUNIX系での例です。
import signal
def handler(signum, frame):
raise TimeoutError("時間切れです")
signal.signal(signal.SIGALRM, handler)
signal.alarm(3) # 3秒後にアラーム発火
try:
time.sleep(10)
except TimeoutError as e:
print(e)
finally:
signal.alarm(0) # アラーム解除
実行結果:
時間切れです
5. まとめ
本記事では、Pythonで一定時間で自動終了するタイムアウト処理の基本と応用を解説しました。
concurrent.futures
によるタイムアウト処理が便利で汎用的- APIアクセスや重い処理にタイムアウトを設けることで、プログラムの安全性と安定性が向上
- Windowsでは
signal
が使えないので、ThreadPoolExecutor
が実用的
実務では「外部サービスの応答がないときにどうするか?」というケースが非常に多いため、今回紹介したタイムアウト処理を覚えておくと非常に役立ちます。
学習のコツとしては、まず小さな処理でタイムアウトの動作を確認し、徐々に大きなプログラムに組み込んでいくと理解しやすくなります。