テクノロジー · 実用ガイド

Jupyter や FastAPI で asyncio.run() が失敗？症状ではなくイベントループの所有者を直す

この例外は所有権の警告です。同じスレッドでホストがすでにループを実行しています。その内部で await するか、本当の同期入口だけで一つのループを所有します。

執筆 Watch at Ease News · 3 分で読めます公開 2026/10/26 3:46 UTC

Python は同じスレッドで二つ目のイベントループを開始する前にエラーを出します。asyncio.run() はループ作成、実行、非同期ジェネレーターの後始末、Executor の終了、ループ閉鎖まで所有します。Notebook カーネルや Web サーバーはすでにその責務を持っています。

安全な修正は呼び出し元で決まります。Notebook セルや非同期ルートでは await し、単独スクリプトでは入口に asyncio.run() を一つだけ置きます。再利用ライブラリは非同期 API を公開し、実行中ループで同期ラッパーが呼ばれたら明確に失敗させます。

変更前にループ所有者を特定する

スレッド、asyncio.get_running_loop() の結果、呼び出したホストを記録します。Notebook、非同期 Web、ライフサイクルフック、CLI は契約が異なります。

自分が作っていないループに stop()、close()、run_until_complete()、asyncio.run() を実行しません。

Jupyter では result = asyncio.run(work()) を result = await work() に変えます。FastAPI では非同期呼び出し元がホストのループで await します。

長時間 Task はホストのライフサイクルに結び付け、開始、キャンセル、終了を定義します。

真の入口に asyncio.run(main()) を置き、共有が必要なら一つの asyncio.Runner で順番に実行します。

ヘルパーごとに runner を作らず、内部を await 可能に保ちます。

主処理を async def fetch_async(...) にし、同期 fetch(...) はループがないときだけ asyncio.run() を使います。ループがあれば await を案内します。

コルーチン作成前に確認し、未使用コルーチンができたなら明示的に閉じます。

別 OS スレッドの専用ループへは asyncio.run_coroutine_threadsafe(coro, loop) を使えますが、時間切れ、キャンセル、終了責任も定義します。

ループスレッドで future.result() を待つとデッドロックし得ます。

必要でも範囲と版を限定したテスト済みアダプターにし、プロセス全体の既定にはしません。

値が返ってもコンテキストエラー、保留タスク、後続セル障害が残る場合があります。

活動中ループの非同期 API と通常スレッドの同期ラッパーを試し、値、例外、キャンセル、contextvars、反復、asyncio.all_tasks() を確認します。

負のテストでは活動中ループ内の同期ラッパーが即座に失敗し、await を案内することを確認します。