【ITニュース解説】Python: Call Multiple APIs in Parallel with asyncio.gather and Semaphore in 10 Minutes

システム開発において、外部のAPIから複数のデータを取得する場面は非常に多く発生する。例えば、多くのユーザー情報をまとめて取得したり、多数の商品の価格情報を並行して収集したりする際、これらのAPI呼び出しを効率的に行う方法がシステムのパフォーマンスを大きく左右する。

もし、これらのAPI呼び出しを一般的な「forループ」を使って一つずつ順番に実行すると、各リクエストが完了するのを待つ時間がすべて累積されてしまう。具体的に言えば、一つのAPIリクエストに1秒かかるとすれば、20個のリクエストは合計で20秒かかり、200個のリクエストであれば3分以上待つことになる。この待ち時間の間、プログラムはAPIからの応答をひたすら待っているだけで、コンピュータの頭脳であるCPUはほとんど何もせずに遊んでいる状態だ。これは、処理の遅延がCPUの計算能力ではなく、データの読み書き（I/O）の待機によって引き起こされる「I/Oバウンド」と呼ばれる状況であり、このような非効率な方法は大規模なデータ処理には適さない。

この深刻な問題を解決するために、Pythonのasyncioライブラリを用いた非同期処理という考え方が非常に有効である。非同期処理の根本的なアイデアは、「ある処理が外部からの応答を待っている間（例えばAPIからのデータ受信を待っている間）、プログラムは時間を無駄にせず、別の処理を開始したり、すでに開始している他の処理を進めたりする」という点にある。これにより、I/O待機時間を有効活用し、複数のAPI呼び出しを「並行して」実行することで、全体の処理時間を劇的に短縮できる。

asyncio.gatherは、複数の非同期タスク（asyncキーワードで定義される「コルーチン」と呼ばれる、一時停止して再開できる小さな処理単位）をまとめて実行し、それらすべてのタスクが完了するのを待ってから、その結果をリストとして一括で返す機能を提供する。このとき、gatherに渡したコルーチンの順番と、返される結果リストの要素の順番は完全に一致するため、どのリクエストがどの結果に対応するかを混乱することなく明確に管理できるのが大きな利点である。

しかし、効率を追求して無制限に多数のAPIリクエストを一度に送ってしまうと、APIを提供しているサーバー側に大きな負担をかける可能性がある。多くのAPIは、短期間に受け付けるリクエスト数に制限（レートリミット）を設けており、この制限を超過するとエラーを返したり、最悪の場合、あなたのプログラムからのアクセスを一時的にブロックしたりすることがある。このような問題を未然に防ぐために利用するのがasyncio.Semaphoreである。Semaphoreは、同時に実行できる非同期処理の数を制限する「ゲート」のような働きをする。例えば、asyncio.Semaphore(5)と設定すれば、同時に実行できるAPIリクエストは最大5つまでに制限される。これにより、APIサーバーに過度な負荷をかけることなく、安全かつ効率的な並行処理を進めることが可能になる。

具体的な実装例として、httpxライブラリの非同期版であるhttpx.AsyncClientを利用し、async defで定義されたfetch_oneというコルーチン内で個々のAPI呼び出しを行う。このfetch_one関数の中でasync with sem:と記述することで、セマフォから実行許可の「チケット」を一時的に取得し、チケットが利用可能な場合にのみAPIリクエストを実行する。そして、APIリクエストが完了するとチケットは自動的に解放され、次に待機している別のAPIリクエストがそのチケットを利用して実行される、という仕組みだ。このようにして、常に設定された数のリクエストだけが並行して実行されるため、安全かつ効率的な処理が実現できる。結果として、20個のリクエストがそれぞれ1秒かかっても、セマフォを5に設定していれば、全体の処理時間は約4秒にまで短縮されることが期待できる。

非同期処理を進める上で、すべてが常に成功するとは限らない。API呼び出しの途中でネットワークエラーやサーバー側の問題などが発生し、一部のリクエストが失敗する可能性は常に考慮すべきである。asyncio.gatherはデフォルトでは、実行中のコルーチンのうち一つでも例外（エラー）が発生すると、すぐにその例外をプログラムに返し、同時に他の実行中のコルーチンの結果も破棄してしまうという挙動をする。これでは、一部の失敗によって、せっかく成功した他の多くのリクエストの結果も無駄になってしまうため、非常に不便である。この問題を解決するには、asyncio.gatherを呼び出す際にreturn_exceptions=Trueという引数を指定する。この設定をすることで、たとえ個々のコルーチン内で例外が発生しても、gatherは処理を中断せずに最後まで実行し、例外オブジェクト自体を成功した結果と同じように結果リストの中に含めて返すようになる。これにより、開発者は後からこの結果リストを調べ、成功した結果と失敗した例外を区別し、失敗したリクエストのみを別途ログに記録したり、リトライ処理を行ったりできる。例えば100件中2件のリクエストが失敗しても、残りの98件は正常に処理を継続できるため、データ同期のバッチ処理などにおいて非常に堅牢なシステムを構築することが可能になる。

実際にプロジェクトでこの非同期処理を導入する際は、まずasyncio.Semaphoreの同時実行数（並行して飛ばすリクエスト数）を5などの少ない値から始め、APIサーバーの応答時間やエラーコード（特にHTTP 429 Too Many Requestsなど）を注意深く監視しながら、徐々にその値を増やしていくのが賢明である。また、すべてのAPIリクエストには必ずタイムアウト時間を設定することが極めて重要だ。もしAPIサーバーが何らかの理由で応答しなくなり、リクエストが永遠に待機状態になった場合、そのリクエストはセマフォの「チケット」を永久に保持し続けることになり、他のすべての待機中のリクエストが実行できなくなって、処理全体が停止してしまう可能性があるからだ。

この非同期処理の技術は、ウェブAPIの呼び出し、データベースへのクエリ、ファイルからのデータ読み込みなど、データの入出力が主な処理時間となる「I/Oバウンド」なタスクで絶大な効果を発揮する。しかし、大量の画像処理や複雑な計算など、CPUが常にフル稼働して計算を行う「CPUバウンド」なタスクには適していない。PythonにはGIL（グローバルインタープリタロック）という仕組みがあり、一つのプロセス内では同時に一つのスレッドしかCPUを利用できないため、純粋な計算処理の並列化にはmultiprocessingモジュールなどの異なるアプローチを検討する必要があることを理解しておくべきだ。