Pythonの async/await で「RuntimeWarning: coroutine was never awaited」を修正する

beginner🐍 Python2026-03-25| Python 3.4以降、すべてのOS(Windows、macOS、Linux)、asyncio または async/await 構文を使用するすべてのプロジェクト

Error Message

RuntimeWarning: coroutine 'xxx' was never awaited
#python#asyncio#coroutine#async#await#runtimewarning

エラーの状況

関数に async def を追加して呼び出すと、Pythonが実行時にこのエラーを投げます:

RuntimeWarning: coroutine 'fetch_data' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

関数は一度も実行されませんでした。データも、副作用も — 何もありません。呼び出しはコルーチンオブジェクトを返し、Pythonはそれを暗黙的に破棄します。

これは非同期Pythonを初めて学ぶほぼ全員がつまずくポイントです。

なぜこうなるのか

async def 関数を呼び出しても、実行はされません。返されるのはコルーチンオブジェクト — メモリ上でスケジューリングを待っている、一時停止した計算処理です。何かがawaitするまで、何も実行されません。

このワーニングが発生する状況は2つあります:

  • 別の非同期関数の中で、awaitなしに非同期関数を呼び出した場合
  • asyncio.run()なしに、通常の同期コードから非同期関数を呼び出した場合

エラーを引き起こす例

import asyncio

async def fetch_data():
    await asyncio.sleep(1)
    return "data"

async def main():
    result = fetch_data()   # バグ: awaitが抜けている
    print(result)           # 出力: <coroutine object fetch_data at 0x...>

asyncio.run(main())

"data"の代わりに、コンソールにコルーチンオブジェクトが表示されます。Pythonはまた次のワーニングも出力します:

RuntimeWarning: coroutine 'fetch_data' was never awaited

修正方法:awaitを追加する

非同期コンテキスト内のすべての非同期関数呼び出しの前に await を置きます。キーワード一つで問題が解決します:

import asyncio

async def fetch_data():
    await asyncio.sleep(1)
    return "data"

async def main():
    result = await fetch_data()   # 修正済み
    print(result)                 # 出力: data

asyncio.run(main())

同期コードから非同期コードを呼び出す

通常の(非同期でない)関数では、awaitは使えません。代わりに asyncio.run() を使いましょう — 新しいイベントループを作成し、コルーチンを完了まで実行して、結果を返します:

import asyncio

async def fetch_data():
    await asyncio.sleep(1)
    return "data"

def main():  # 通常の同期関数
    result = asyncio.run(fetch_data())   # 正しい方法
    print(result)

main()

**重要な注意点:**すでに実行中のイベントループの中で asyncio.run() を呼び出してはいけません — 例えば、別の非同期関数の中、Jupyterノートブックのセル、またはFastAPIのリクエストハンドラーの中などです。別のエラーが発生します:

RuntimeError: This event loop is already running

そのような環境では、直接 await を使いましょう。

複数のコルーチンを並行実行する

複数の非同期関数を一度に起動したいですか?asyncio.gather() を使いましょう。すべてを並行実行して、結果をまとめて返します:

import asyncio

async def task_a():
    await asyncio.sleep(1)
    return "A done"

async def task_b():
    await asyncio.sleep(1)
    return "B done"

async def main():
    # 両方のタスクが並行実行される — 合計時間は約1秒(2秒ではなく)
    results = await asyncio.gather(task_a(), task_b())
    print(results)  # ['A done', 'B done']

asyncio.run(main())

awaitを省略すると振り出しに戻ります — 2つのコルーチンオブジェクトが作成されるだけで、実行されず、ワーニングが発生します。

クラスメソッドの落とし穴

Pythonの __init__ メソッドは async にできません。非同期初期化が必要なオブジェクトをセットアップする際に、これで戸惑う人が多いです:

class DataFetcher:
    async def load(self):
        await asyncio.sleep(0.5)
        self.data = "loaded"

# 誤り — __init__はasyncにできないため、ここではawaitできない
class App:
    def __init__(self):
        fetcher = DataFetcher()
        fetcher.load()  # コルーチンがawaitされない!

# 正しい — 非同期セットアップを専用の非同期メソッドに移す
class App:
    async def setup(self):
        fetcher = DataFetcher()
        await fetcher.load()
        self.fetcher = fetcher

async def main():
    app = App()
    await app.setup()

asyncio.run(main())

tracemallocで正確な行を特定する

大規模なコードベースでは、ワーニングだけではどこでawaitされていないコルーチンが作成されたかわかりません。スクリプトの先頭で tracemalloc を有効にすると、Pythonが正確なファイルと行番号を含めて表示します:

import tracemalloc
tracemalloc.start()

import asyncio

async def fetch_data():
    return "data"

async def main():
    fetch_data()  # awaitされていない

asyncio.run(main())

ワーニングの出力にメモリ確保のトレースバックが表示されるようになります — 何百行ものコードを調べるよりずっと便利です。

開発中はハードエラーにする

ワーニングは見落としやすく、特にテスト出力に埋もれている場合はなおさらです。これをハードクラッシュに昇格させて、大きな音で失敗するようにしましょう:

import warnings
warnings.filterwarnings("error", category=RuntimeWarning)

またはPythonに直接フラグを渡します:

python -W error::RuntimeWarning your_script.py

これで、awaitされていないコルーチンはすぐに例外を発生させ、実行を停止します。CIが本番環境に届く前に検出してくれます。

修正を確認する

  • スクリプトを再実行します。RuntimeWarningの行が消えているはずです。
  • 非同期関数の戻り値が <coroutine object ...> という文字列ではなく、実際のデータであることを確認します。
  • テストスイートに簡単なアサーションを追加します:
import asyncio

async def fetch_data():
    return "data"

async def test_fetch():
    result = await fetch_data()
    assert result == "data", f"Expected 'data', got {result!r}"
    print("Test passed")

asyncio.run(test_fetch())

クイックリファレンス

  • async → async: result = await func()
  • sync → async: result = asyncio.run(func())
  • 複数の並行タスク: results = await asyncio.gather(func1(), func2())
  • どの行かわからない? スクリプトの先頭に tracemalloc.start() を追加する
  • CI/テストで: -W error::RuntimeWarning オプションで実行してハードクラッシュにする

Related Error Notes