コルーチンとタスク

この節では、コルーチンと Task を利用する高レベルの asyncio の API の概略を解説します。

コルーチン

ソースコード: Lib/asyncio/coroutines.py

---

async/await 構文で宣言された コルーチン は、 asyncio を使ったアプリケーションを書くのに推奨される方法です。例えば、次のコードスニペットは "hello" を出力し、そこから 1 秒待って "world" を出力します:

>>> import asyncio

>>> async def main():
...     print('hello')
...     await asyncio.sleep(1)
...     print('world')

>>> asyncio.run(main())
hello
world

単にコルーチンを呼び出しただけでは、コルーチンの実行スケジュールは予約されていないことに注意してください:

>>> main()
<coroutine object main at 0x1053bb7c8>

実際にコルーチンを実行するために、asyncio は以下のメカニズムを提供しています:

• 最上位のエントリーポイントである "main()" 関数を実行する asyncio.run() 関数 (上の例を参照してください。)

• コルーチンを await すること。次のコード片は 1 秒間待機した後に "hello" と出力し、 更に 2 秒間待機してから "world" と出力します:

  import asyncio
  import time
  
  async def say_after(delay, what):
      await asyncio.sleep(delay)
      print(what)
  
  async def main():
      print(f"started at {time.strftime('%X')}")
  
      await say_after(1, 'hello')
      await say_after(2, 'world')
  
      print(f"finished at {time.strftime('%X')}")
  
  asyncio.run(main())
  

  予想される出力:

  started at 17:13:52
  hello
  world
  finished at 17:13:55
  

• asyncio の Tasks としてコルーチンを並行して走らせる asyncio.create_task() 関数。

  上のコード例を編集して、ふたつの say_after コルーチンを 並行して 走らせてみましょう:

  async def main():
      task1 = asyncio.create_task(
          say_after(1, 'hello'))
  
      task2 = asyncio.create_task(
          say_after(2, 'world'))
  
      print(f"started at {time.strftime('%X')}")
  
      # 両方のタスクが完了するまで待つ (約 2 秒かかる)。
      await task1
      await task2
  
      print(f"finished at {time.strftime('%X')}")
  

  予想される出力が、スニペットの実行が前回よりも 1 秒早いことを示していることに注意してください:

  started at 17:14:32
  hello
  world
  finished at 17:14:34
  

• asyncio.TaskGroup クラスは、create_task() に代わるより新しい手段です。 この API を使うと、直前の例は次のようになります。

  async def main():
      async with asyncio.TaskGroup() as tg:
          task1 = tg.create_task(
              say_after(1, 'hello'))
  
          task2 = tg.create_task(
              say_after(2, 'world'))
  
          print(f"started at {time.strftime('%X')}")
  
      # コンテキストマネージャーを抜けるときに暗黙に await する。
  
      print(f"finished at {time.strftime('%X')}")
  

  実行にかかる時間と出力は、直前の例と同じになるはずです。

  Added in version 3.11: asyncio.TaskGroup.

Awaitable

あるオブジェクトを await 式の中で使うことができる場合、そのオブジェクトを awaitable オブジェクトと言います。多くの asyncio API は awaitable を受け取るように設計されています。

awaitable オブジェクトには主に3つの種類があります: コルーチン, Task, そして Future です

コルーチン

Python のコルーチンは awaitable であり、他のコルーチンから待機されることができます:

import asyncio

async def nested():
    return 42

async def main():
    # "nested()" を呼び出すだけでは何も起こらない。
    # コルーチンオブジェクトは作成されるが await されないので、
    # まったく実行されない。
    nested()  # "RuntimeWarning" が発生する。

    # 今度は await してみる:
    print(await nested())  # "42" と表示する。

asyncio.run(main())

重要

このドキュメントにおいて「コルーチン」という用語は以下2つの密接に関連した概念に対して使用できます:

• コルーチン関数: async def 関数;

• コルーチンオブジェクト: コルーチン関数 を呼び出すと返ってくるオブジェクト.

Task

Task は、コルーチンを 並行に スケジュールするのに使います。

asyncio.create_task() のような関数で、コルーチンが Task にラップされているとき、自動的にコルーチンは即時実行されるようにスケジュールされます:

import asyncio

async def nested():
    return 42

async def main():
    # nested() がまもなく "main()" と並行して実行されるようにする。
    task = asyncio.create_task(nested())

    # "task" で "nested()" をキャンセルしたり、
    # 単に await して完了を待ったりできる:
    await task

asyncio.run(main())

Future

Future は、非同期処理の 最終結果 を表現する特別な 低レベルの awaitable オブジェクトです。

Future オブジェクトが 待機 (await) されている とは、Future がどこか他の場所で解決されるまでコルーチンが待機するということです。

asyncioのFutureオブジェクトを使うと、async/awaitとコールバック形式のコードを併用できます。

通常、アプリケーション水準のコードで Future オブジェクトを作る 必要はありません 。

Future オブジェクトはライブラリや asyncio のAPIで外部に提供されることもあり、await （待機）されることができます:

async def main():
    await function_that_returns_a_future_object()

    # これも有効:
    await asyncio.gather(
        function_that_returns_a_future_object(),
        some_python_coroutine()
    )

Future オブジェクトを返す低レベル関数の良い例は loop.run_in_executor() です。

タスクの作成

ソースコード: Lib/asyncio/tasks.py

---

asyncio.create_task(coro, *, name=None, context=None, eager_start=None, **kwargs)

coro coroutine を Task でラップし、その実行をスケジュールします。 Task オブジェクトを返します。

関数のシグネチャ全体は Task のコンストラクター (またはファクトリー) とほぼ同じです。 この関数に渡したキーワード引数はすべて、そのコンストラクターやファクトリーに渡されます。

省略可能なキーワード引数 context によって、coro を実行するためのカスタムの contextvars.Context を指定できます。context が省略された場合、現在のコンテキストのコピーが作成されます。

省略可能なキーワード専用引数 eager_start では、create_task() の呼び出し中にタスクの実行を即座に開始するか、後で実行するようにスケジュールするかを指定できます。 eager_start を渡さない場合は loop.set_task_factory() で設定したモードを使います。

その Task オブジェクトは get_running_loop() から返されたループの中で実行されます。現在のスレッドに実行中のループが無い場合は、 RuntimeError が送出されます。

注釈

asyncio.TaskGroup.create_task() は、構造化された並行処理を使う新しい代替手段です。 強い安全性の保証のもとで、関連するタスクのグループを待機できます。

重要

タスクが実行中に消えないように、この関数の結果の参照を保存してください。イベントループは弱い参照のみを保持します。ほかに参照元のないタスクは、完了していなくてもガーベジコレクションされる可能性があります。信頼性のある "fire-and-forget" バックグラウンドタスクが必要な場合、コレクションを使ってください。

background_tasks = set()

for i in range(10):
    task = asyncio.create_task(some_coro(param=i))

    # タスクを集合に追加する。これにより強参照が作成される。
    background_tasks.add(task)

    # 完了したタスクへの参照を保持し続けないように、
    # 各タスクが完了したら集合から自身への参照を削除するようにする:
    task.add_done_callback(background_tasks.discard)

この方法ではタスクを await しないため、タスクが失敗しても例外は取得されません。 タスクがガベージコレクションされると、asyncio は "Task exception was never retrieved" というメッセージをログに記録します。 これを避けるには asyncio.TaskGroup を使ってください。 各タスクへの強参照を保持してタスクを await し、例外を伝播させます。

async with asyncio.TaskGroup() as tg:
    for i in range(10):
        tg.create_task(some_coro(param=i))

Added in version 3.7.

バージョン 3.8 で変更: name パラメータを追加しました。

バージョン 3.11 で変更: context パラメータを追加しました。

バージョン 3.14 で変更: すべての kwargs を渡すようにすることで、eager_start パラメーターが追加されました。

タスクのキャンセル

タスクは簡単に、そして安全にキャンセルできます。タスクがキャンセルされた場合、asyncio.CancelledError が次の機会に送出されます。

後片付けを確実に行うため、コルーチンでは try/finally ブロックを使うことを推奨します。 asyncio.CancelledError を明示的に捕捉した場合、通常は後片付けが終わったら伝播させてください。 asyncio.CancelledError は BaseException の直接のサブクラスなので、ほとんどのコードでは特別に意識する必要はありません。

asyncio.TaskGroup や asyncio.timeout() など、構造化された並行処理を実現する asyncio の構成要素は内部でキャンセルを使って実装されています。 コルーチンが asyncio.CancelledError を握りつぶすと、正しく動作しない場合があります。 同様に、ユーザーのコードから uncancel を呼び出すことも通常は避けてください。 ただし、本当に asyncio.CancelledError を抑制したい場合は、キャンセル状態を完全に解除するために uncancel() も呼び出す必要があります。

タスクグループ

タスクグループは、タスクを作成する API と、グループ内のすべてのタスクの終了を便利かつ確実に待つ手段を組み合わせたものです。

class asyncio.TaskGroup

タスクのグループを保持する 非同期コンテキストマネージャー です。 create_task() でタスクをグループに追加できます。 コンテキストマネージャーを抜けるときに、すべてのタスクを await します。

Added in version 3.11.

create_task(coro, *, name=None, context=None, eager_start=None, **kwargs)

このタスクグループにタスクを作成します。 シグネチャは asyncio.create_task() と同じです。 タスクグループが非アクティブ (まだ開始していない、すでに終了した、終了処理中など) の場合は、渡された coro を閉じて RuntimeError を送出します。

バージョン 3.13 で変更: タスクグループがアクティブでない場合、渡されたコルーチンを閉じるようになりました。

バージョン 3.14 で変更: すべての kwargs を loop.create_task() に渡すようになりました。

以下はプログラム例です:

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(some_coro(...))
        task2 = tg.create_task(another_coro(...))
    print(f"Both tasks have completed now: {task1.result()}, {task2.result()}")

The async with statement will wait for all tasks in the group to finish. While waiting, new tasks may still be added to the group (for example, by passing tg into one of the coroutines and calling tg.create_task() in that coroutine). Once the last task has finished and the async with block is exited, no new tasks may be added to the group.

グループ内のいずれかのタスクが asyncio.CancelledError 以外の例外で初めて失敗した時点で、グループ内の残りのタスクをキャンセルします。 それ以降はグループにタスクを追加できません。 この時点で async with 文の本体がまだ実行中 (つまり __aexit__() がまだ呼び出されていない) なら、async with 文を直接含むタスクもキャンセルされます。 これによって生じる asyncio.CancelledError は await を中断しますが、それを含む async with 文の外には伝播しません。

すべてのタスクが終了した時点で、asyncio.CancelledError 以外の例外で失敗したタスクがあれば、それらの例外を ExceptionGroup または BaseExceptionGroup にまとめて送出します (どちらを使うかは、それぞれのドキュメントを参照してください)。

基底例外のうち、2 つは特別に扱います。 いずれかのタスクが KeyboardInterrupt または SystemExit で失敗した場合も、タスクグループは残りのタスクをキャンセルして終了を待ちます。 ただし、その後は ExceptionGroup や BaseExceptionGroup ではなく、最初の KeyboardInterrupt または SystemExit を再送出します。

If the body of the async with statement exits with an exception (so __aexit__() is called with an exception set), this is treated the same as if one of the tasks failed: the remaining tasks are cancelled and then waited for, and non-cancellation exceptions are grouped into an exception group and raised. The exception passed into __aexit__(), unless it is asyncio.CancelledError, is also included in the exception group. The same special case is made for KeyboardInterrupt and SystemExit as in the previous paragraph.

タスクグループは、__aexit__() を「起こす」ための内部的なキャンセルと、タスクグループが実行されているタスクに対する外部からのキャンセル要求を区別します。 特に、タスクグループが構文上入れ子になっていて、両方のグループの子タスクで同時に例外が発生した場合は、まず内側のグループが例外を処理します。 その後、外側のグループが別のキャンセルを受け取り、自身の例外を処理します。

タスクグループが外部からキャンセルされ、かつ ExceptionGroup を送出する必要もある場合は、親タスクの cancel() メソッドを呼び出します。 これにより、次の await で asyncio.CancelledError が送出され、キャンセルが失われないようにします。

タスクグループは asyncio.Task.cancelling() が返すキャンセル回数を保持します。

バージョン 3.13 で変更: 内部と外部からのキャンセルが同時に発生した場合の処理が改善され、キャンセル回数が正しく保持されるようになりました。

Terminating a task group

While terminating a task group is not natively supported by the standard library, termination can be achieved by adding an exception-raising task to the task group and ignoring the raised exception:

import asyncio
from asyncio import TaskGroup

class TerminateTaskGroup(Exception):
    """Exception raised to terminate a task group."""

async def force_terminate_task_group():
    """Used to force termination of a task group."""
    raise TerminateTaskGroup()

async def job(task_id, sleep_time):
    print(f'Task {task_id}: start')
    await asyncio.sleep(sleep_time)
    print(f'Task {task_id}: done')

async def main():
    try:
        async with TaskGroup() as group:
            # spawn some tasks
            group.create_task(job(1, 0.5))
            group.create_task(job(2, 1.5))
            # sleep for 1 second
            await asyncio.sleep(1)
            # add an exception-raising task to force the group to terminate
            group.create_task(force_terminate_task_group())
    except* TerminateTaskGroup:
        pass

asyncio.run(main())

Expected output:

Task 1: start
Task 2: start
Task 1: done

スリープ

async asyncio.sleep(delay, result=None)

delay 秒だけ停止します。

result が提供されている場合は、コルーチン完了時にそれが呼び出し元に返されます。

sleep() は常に現在の Task を一時中断し、他の Task が実行されるのを許可します。

delay を 0 に設定すると、最適化された処理経路で他のタスクを実行できるようにします。 長時間実行する関数でこれを使うと、関数の呼び出し中ずっとイベントループをブロックしてしまうのを避けられます。

現在の時刻を5秒間、毎秒表示するコルーチンの例:

import asyncio
import datetime as dt

async def display_date():
    loop = asyncio.get_running_loop()
    end_time = loop.time() + 5.0
    while True:
        print(dt.datetime.now())
        if (loop.time() + 1.0) >= end_time:
            break
        await asyncio.sleep(1)

asyncio.run(display_date())

バージョン 3.10 で変更: loop パラメータが削除されました。

バージョン 3.13 で変更: delay が nan の場合に ValueError を送出するようになりました。

タスクの並行実行

awaitable asyncio.gather(*aws, return_exceptions=False)

aws シーケンスにある awaitable オブジェクト を 並行 実行します。

aws にある awaitable がコルーチンである場合、自動的に Task としてスケジュールされます。

全ての awaitable が正常終了した場合、その結果は返り値を集めたリストになります。 返り値の順序は、 aws での awaitable の順序に相当します。

return_exceptions が False である場合(デフォルト)、gather() で await しているタスクに対して、最初の例外が直接伝えられます。aws に並んでいる他の awaitable は、キャンセルされずに 引き続いて実行されます。

return_exceptions が True だった場合、例外は成功した結果と同じように取り扱われ、結果リストに集められます。

gather() が キャンセル された場合、起動された全ての (未完了の) awaitable も キャンセル されます。

aws シーケンスにある Task あるいは Future が キャンセル された場合、 CancelledError を送出したかのうように扱われます。つまり、この場合 gather() 呼び出しはキャンセル されません。 これは、起動された 1 つの Task あるいは Future のキャンセルが、他の Task あるいは Future のキャンセルを引き起こすのを避けるためです。

注釈

タスクを作成して並行に実行し、完了を待つための新しい代替手段として asyncio.TaskGroup があります。 サブタスクを入れ子にスケジュールする場合、TaskGroup は gather よりも強い安全性を保証します。 タスク (またはタスクがスケジュールしたサブタスク) が例外を送出すると、TaskGroup はスケジュール済みの残りのタスクをキャンセルしますが、gather はキャンセルしません。

以下はプログラム例です:

import asyncio

async def factorial(name, number):
    f = 1
    for i in range(2, number + 1):
        print(f"Task {name}: Compute factorial({number}), currently i={i}...")
        await asyncio.sleep(1)
        f *= i
    print(f"Task {name}: factorial({number}) = {f}")
    return f

async def main():
    # 3 つの呼び出しを並行に実行するようスケジュールする:
    L = await asyncio.gather(
        factorial("A", 2),
        factorial("B", 3),
        factorial("C", 4),
    )
    print(L)

asyncio.run(main())

# 期待される出力:
#
#     Task A: Compute factorial(2), currently i=2...
#     Task B: Compute factorial(3), currently i=2...
#     Task C: Compute factorial(4), currently i=2...
#     Task A: factorial(2) = 2
#     Task B: Compute factorial(3), currently i=3...
#     Task C: Compute factorial(4), currently i=3...
#     Task B: factorial(3) = 6
#     Task C: Compute factorial(4), currently i=4...
#     Task C: factorial(4) = 24
#     [2, 6, 24]

注釈

return_exceptions が偽の場合、完了状態になった後で gather() をキャンセルしても、渡された awaitable はキャンセルされません。 例えば、gather は例外を呼び出し元に伝播した時点で完了状態になることがあります。 そのため、いずれかの awaitable が送出した例外を gather から捕捉した後で gather.cancel() を呼び出しても、他の awaitable はキャンセルされません。

バージョン 3.7 で変更: gather 自身がキャンセルされた場合は、 return_exceptions の値に関わらずキャンセルが伝搬されます。

バージョン 3.10 で変更: loop パラメータが削除されました。

バージョン 3.10 で非推奨: 位置引数が渡されていない場合、または位置引数のすべてが Future のようなオブジェクトというわけではなく、かつ実行中のイベントループがない場合に、非推奨の警告を出すようになりました。

即時実行するタスクファクトリー

asyncio.eager_task_factory(loop, coro, *, name=None, context=None)

タスクの実行を即座に開始するタスクファクトリーです。

このファクトリーを loop.set_task_factory(asyncio.eager_task_factory) で使うと、Task の構築中にコルーチンの実行を同期的に開始します。 タスクがイベントループにスケジュールされるのは、処理がブロックした場合だけです。 同期的に完了するコルーチンではループへのスケジュールに伴うオーバーヘッドを避けられるため、性能が向上する場合があります。

これが役立つ典型例は、キャッシュやメモ化を使って可能な限り実際の I/O を避けるコルーチンです。

注釈

コルーチンを即座に実行すると、実行の意味が変わります。 コルーチンが値を返すか例外を送出すると、タスクはイベントループにスケジュールされません。 コルーチンの実行がブロックすると、タスクはイベントループにスケジュールされます。 この変更により、既存のアプリケーションの動作が変わる可能性があります。 例えば、タスクの実行順序は変わる可能性が高いでしょう。

Added in version 3.12.

asyncio.create_eager_task_factory(custom_task_constructor)

eager_task_factory() と同様に即時実行するタスクファクトリーを作成します。 新しいタスクの作成には、デフォルトの Task の代わりに、指定された custom_task_constructor を使います。

custom_task_constructor は Task.__init__ と同じシグネチャを持つ 呼び出し可能オブジェクト でなければなりません。 このオブジェクトは asyncio.Task と互換性のあるオブジェクトを返す必要があります。

この関数は、loop.set_task_factory(factory) でイベントループのタスクファクトリーとして使うための 呼び出し可能オブジェクト を返します。

Added in version 3.12.

キャンセルからの保護

awaitable asyncio.shield(aw)

キャンセル から awaitable オブジェクト を保護します。

aw がコルーチンだった場合、自動的に Task としてスケジュールされます。

文:

task = asyncio.create_task(something())
res = await shield(task)

は、以下と同じです

res = await something()

ただし、これを含むコルーチンがキャンセルされても、something() を実行している Task はキャンセルされない という違いがあります。 something() 側から見ると、キャンセルは発生していません。 呼び出し元はキャンセルされるため、"await" 式は CancelledError を送出します。

注意: something() が他の理由 (例えば、原因が自分自身) でキャンセルされた場合は shield() でも保護できません。

完全にキャンセルを無視したい場合 (推奨はしません) は、 shield() 関数は次のように try/except 節と組み合わせることになるでしょう:

task = asyncio.create_task(something())
try:
    res = await shield(task)
except CancelledError:
    res = None

重要

タスクが実行途中で消えないように、この関数に渡すタスクへの参照を保持してください。 イベントループはタスクへの弱参照しか保持しません。 他から参照されていないタスクは、完了前でもいつでもガベージコレクションされる可能性があります。

バージョン 3.10 で変更: loop パラメータが削除されました。

バージョン 3.10 で非推奨: aw が Future のようなオブジェクトでなく、実行中のイベントループもない場合に、非推奨の警告を出すようになりました。

タイムアウト

asyncio.timeout(delay)

待機に費やす時間を制限できる 非同期コンテキストマネージャー を返します。

delay には None、または待機する秒数を浮動小数点数か整数で指定できます。 delay が None の場合、時間制限は適用されません。 コンテキストマネージャーの作成時点で待機時間が分からない場合に便利です。

どちらの場合も、作成後に Timeout.reschedule() でコンテキストマネージャーの期限を変更できます。

以下はプログラム例です:

async def main():
    async with asyncio.timeout(10):
        await long_running_task()

long_running_task の完了に 10 秒を超えてかかると、コンテキストマネージャーは現在のタスクをキャンセルします。 その結果生じる asyncio.CancelledError を内部で処理し、捕捉して処理できる TimeoutError に変換します。

注釈

asyncio.CancelledError を TimeoutError に変換するのは asyncio.timeout() のコンテキストマネージャーです。 そのため、TimeoutError を捕捉できるのはコンテキストマネージャーの 外側 だけです。

TimeoutError を捕捉する例です。

async def main():
    try:
        async with asyncio.timeout(10):
            await long_running_task()
    except TimeoutError:
        print("The long operation timed out, but we've handled it.")

    print("This statement will run regardless.")

asyncio.timeout() が生成するコンテキストマネージャーは、期限を変更したり、状態を調べたりできます。

class asyncio.Timeout(when)

期限を超えたコルーチンをキャンセルするための 非同期コンテキストマネージャー です。

Timeout を直接インスタンス化するよりも、asyncio.timeout() または asyncio.timeout_at() を使うことを推奨します。

when には、イベントループの時計で計測した、タイムアウトすべき絶対時刻を指定してください。

• when が None の場合、タイムアウトは発生しません。

• when < loop.time() の場合、イベントループの次の反復でタイムアウトが発生します。

when() → float | None

現在の期限を返します。 期限が設定されていなければ None を返します。

reschedule(when: float | None)

タイムアウトの期限を変更します。

expired() → bool

コンテキストマネージャーが期限を過ぎたかどうかを返します。

以下はプログラム例です:

async def main():
    try:
        # 開始時には期限が分からないので、``None`` を渡す。
        async with asyncio.timeout(None) as cm:
            # 期限が分かったので設定する。
            new_deadline = get_running_loop().time() + 10
            cm.reschedule(new_deadline)

            await long_running_task()
    except TimeoutError:
        pass

    if cm.expired():
        print("Looks like we haven't finished on time.")

タイムアウトのコンテキストマネージャーは安全に入れ子にできます。

Added in version 3.11.

asyncio.timeout_at(when)

asyncio.timeout() と似ていますが、when には待機を終了する絶対時刻、または None を指定します。

以下はプログラム例です:

async def main():
    loop = get_running_loop()
    deadline = loop.time() + 20
    try:
        async with asyncio.timeout_at(deadline):
            await long_running_task()
    except TimeoutError:
        print("The long operation timed out, but we've handled it.")

    print("This statement will run regardless.")

Added in version 3.11.

async asyncio.wait_for(aw, timeout)

aw awaitable が、完了するかタイムアウトになるのを待ちます。

timeout には None もしくは待つ秒数の浮動小数点数か整数を指定できます。 timeout が None の場合、 Future が完了するまで待ちます。

タイムアウトが発生すると aw をキャンセルし、TimeoutError を送出します。

aw がキャンセルされないようにするには shield() でラップしてください。

この関数は future が実際にキャンセルされるまで待つため、待ち時間の合計は timeout を超えることがあります。キャンセル中に例外が発生した場合は、その例外は伝達されます。

待機が中止された場合 aw も中止されます。

以下はプログラム例です:

async def eternity():
    # 1 時間スリープする
    await asyncio.sleep(3600)
    print('yay!')

async def main():
    # 最大 1 秒待つ
    try:
        await asyncio.wait_for(eternity(), timeout=1.0)
    except TimeoutError:
        print('timeout!')

asyncio.run(main())

# 期待される出力:
#
#     timeout!

バージョン 3.7 で変更: タイムアウトにより aw をキャンセルする場合、wait_for は aw のキャンセルが完了するまで待つようになりました。 以前は TimeoutError を即座に送出していました。

バージョン 3.10 で変更: loop パラメータが削除されました。

バージョン 3.11 で変更: asyncio.TimeoutError の代わりに TimeoutError を送出するようになりました。

バージョン 3.12 で変更: asyncio.timeout() を使った実装になり、timeout が正の場合に aw として渡されたコルーチンを Task でラップしなくなりました。

待機プリミティブ

async asyncio.wait(aws, *, timeout=None, return_when=ALL_COMPLETED)

イテラブル aws に含まれる Future と Task のインスタンスを並行に実行し、return_when で指定した条件が満たされるまでブロックします。

イテラブル aws は空であってはなりません。

Task/Future からなる 2 つの集合 (done, pending) を返します。

使い方:

done, pending = await asyncio.wait(aws)

timeout (浮動小数点数または整数) が指定されていたら、処理を返すのを待つ最大秒数を制御するのに使われます。

この関数は TimeoutError を送出しません。 タイムアウト時点で完了していない Future や Task は、単に 2 番目の集合に入れて返されます。

return_when でこの関数がいつ結果を返すか指定します。指定できる値は以下の 定数のどれか一つです:

定数	説明
asyncio.FIRST_COMPLETED	いずれかのフューチャが終了したかキャンセルされたときに返します。
asyncio.FIRST_EXCEPTION	いずれかのフューチャが例外を送出して終了すると、この関数は返ります。 例外を送出するフューチャがない場合は、ALL_COMPLETED と等価です。
asyncio.ALL_COMPLETED	すべてのフューチャが終了したかキャンセルされたときに返します。

wait_for() と異なり、 wait() はタイムアウトが起きたときに Future をキャンセルしません。

wait() がキャンセルされても、aws 内の Future はキャンセルされずに実行を続けます。

バージョン 3.10 で変更: loop パラメータが削除されました。

バージョン 3.11 で変更: wait() にコルーチンオブジェクトを直接渡すことは禁止されました。

バージョン 3.12 で変更: タスクを yield するジェネレーターをサポートするようになりました。

asyncio.as_completed(aws, *, timeout=None)

イテラブル aws に含まれる awaitable オブジェクト を並行に実行します。 返されたオブジェクトを反復すると、awaitable の完了に応じて結果を取得できます。

as_completed() が返すオブジェクトは asynchronous iterator としても通常の iterator としても反復できます。 非同期反復の場合、もともと渡された awaitable が Task または Future なら、そのオブジェクトを yield します。 これにより、スケジュール済みのタスクと結果を簡単に対応付けられます。 例えば次のように使います。

ipv4_connect = create_task(open_connection("127.0.0.1", 80))
ipv6_connect = create_task(open_connection("::1", 80))
tasks = [ipv4_connect, ipv6_connect]

async for earliest_connect in as_completed(tasks):
    # earliest_connect は完了している。await するか
    # earliest_connect.result() を呼び出すと結果を取得できる。
    reader, writer = await earliest_connect

    if earliest_connect is ipv6_connect:
        print("IPv6 connection established.")
    else:
        print("IPv4 connection established.")

非同期反復では、渡された awaitable が Task や Future でない場合、暗黙に作成されたタスクを yield します。

通常のイテレーターとして使う場合、反復のたびに新しいコルーチンを yield します。 このコルーチンは、次に完了する awaitable の結果を返すか、その例外を送出します。 この使い方は Python 3.13 より前のバージョンと互換性があります。

ipv4_connect = create_task(open_connection("127.0.0.1", 80))
ipv6_connect = create_task(open_connection("::1", 80))
tasks = [ipv4_connect, ipv6_connect]

for next_connect in as_completed(tasks):
    # next_connect は元のタスクオブジェクトではない。
    # await すると、次に完了する awaitable の結果を取得するか、
    # その例外を送出する。
    reader, writer = await next_connect

すべての awaitable が完了する前にタイムアウトすると、TimeoutError を送出します。 非同期反復では async for ループが、通常の反復では yield されたコルーチンが、この例外を送出します。

as_completed() は、渡された awaitable を実行しているタスクをキャンセルしません。 タイムアウトした場合や反復がキャンセルされた場合も、残りのタスクは実行を続けます。

バージョン 3.10 で変更: loop パラメータが削除されました。

バージョン 3.10 で非推奨: イテラブル aws に含まれる awaitable オブジェクトのすべてが Future のようなオブジェクトというわけではなく、かつ実行中のイベントループがない場合に、非推奨の警告を出すようになりました。

バージョン 3.12 で変更: タスクを yield するジェネレーターをサポートするようになりました。

バージョン 3.13 で変更: 結果を asynchronous iterator としても通常の iterator としても使えるようになりました (以前は通常のイテレーターとしてのみ使えました)。

スレッドでの実行

async asyncio.to_thread(func, /, *args, **kwargs)

別のスレッドで非同期的に関数 func を実行します。

この関数に渡された *args と **kwargs は関数 func に直接渡されます。また、イベントループスレッドのコンテキスト変数に関数を実行するスレッドからアクセスできるように、現在の contextvars.Context も伝播されます。

関数 func の最終結果を待ち受けできるコルーチンを返します。

このコルーチン関数は主に、メインスレッドで実行するとイベントループをブロックしてしまう I/O バウンドの関数やメソッドを実行するためのものです。 例えば次のように使います。

def blocking_io():
    print(f"start blocking_io at {time.strftime('%X')}")
    # time.sleep() は、ファイル操作などのブロックする
    # I/O バウンドの処理に置き換えられる。
    time.sleep(1)
    print(f"blocking_io complete at {time.strftime('%X')}")

async def main():
    print(f"started main at {time.strftime('%X')}")

    await asyncio.gather(
        asyncio.to_thread(blocking_io),
        asyncio.sleep(1))

    print(f"finished main at {time.strftime('%X')}")


asyncio.run(main())

# 期待される出力:
#
# started main at 19:50:53
# start blocking_io at 19:50:53
# blocking_io complete at 19:50:54
# finished main at 19:50:54

コルーチン内で blocking_io() を直接呼び出すと、実行中ずっとイベントループをブロックするため、実行時間が 1 秒長くなります。 代わりに asyncio.to_thread() を使えば、イベントループをブロックせずに別のスレッドで実行できます。

注釈

Due to the GIL, asyncio.to_thread() can typically only be used to make IO-bound functions non-blocking. However, for extension modules that release the GIL or alternative Python implementations that don't have one, asyncio.to_thread() can also be used for CPU-bound functions.

Added in version 3.9.

他のスレッドからのスケジュール

asyncio.run_coroutine_threadsafe(coro, loop)

与えられたイベントループにコルーチンを送ります。 この処理は、スレッドセーフです。

他の OS スレッドから結果を待つための concurrent.futures.Future を返します。

この関数は、イベントループが動作しているスレッドとは異なる OS スレッドから呼び出すためのものです。 例えば次のように使います:

def in_thread(loop: asyncio.AbstractEventLoop) -> None:
    # ブロックする I/O を実行する
    pathlib.Path("example.txt").write_text("hello world", encoding="utf8")

    # コルーチンを作成する
    coro = asyncio.sleep(1, result=3)

    # 指定されたループにコルーチンを送る
    future = asyncio.run_coroutine_threadsafe(coro, loop)

    # 省略可能な timeout 引数を指定して結果を待つ
    assert future.result(timeout=2) == 3

async def amain() -> None:
    # 実行中のループを取得する
    loop = asyncio.get_running_loop()

    # スレッドで処理を実行する
    await asyncio.to_thread(in_thread, loop)

逆に、別のスレッドで実行するイベントループへ処理を送ることもできます。 例えば次のようにします。

@contextlib.contextmanager
def loop_in_thread() -> Generator[asyncio.AbstractEventLoop]:
    loop_fut = concurrent.futures.Future[asyncio.AbstractEventLoop]()
    stop_event = asyncio.Event()

    async def main() -> None:
        loop_fut.set_result(asyncio.get_running_loop())
        await stop_event.wait()

    with concurrent.futures.ThreadPoolExecutor(1) as tpe:
        complete_fut = tpe.submit(asyncio.run, main())
        for fut in concurrent.futures.as_completed((loop_fut, complete_fut)):
            if fut is loop_fut:
                loop = loop_fut.result()
                try:
                    yield loop
                finally:
                    loop.call_soon_threadsafe(stop_event.set)
            else:
                fut.result()

# 別のスレッドにループを作成する
with loop_in_thread() as loop:
    # コルーチンを作成する
    coro = asyncio.sleep(1, result=3)

    # 指定されたループにコルーチンを送る
    future = asyncio.run_coroutine_threadsafe(coro, loop)

    # 省略可能な timeout 引数を指定して結果を待つ
    assert future.result(timeout=2) == 3

コルーチンから例外が送出された場合、返された Future に通知されます。 これはイベントループの Task をキャンセルするのにも使えます:

try:
    result = future.result(timeout)
except TimeoutError:
    print('The coroutine took too long, cancelling the task...')
    future.cancel()
except Exception as exc:
    print(f'The coroutine raised an exception: {exc!r}')
else:
    print(f'The coroutine returned: {result!r}')

このドキュメントの 並行処理とマルチスレッド処理 節を参照してください。

他の asyncio 関数とは異なり、この関数は明示的に渡される loop 引数を必要とします。

Added in version 3.5.1.

イントロスペクション

asyncio.current_task(loop=None)

現在実行中の Task インスタンスを返します。実行中の Task が無い場合は None を返します。

loop が None の場合、 get_running_loop() が現在のループを取得するのに使われます。

Added in version 3.7.

asyncio.all_tasks(loop=None)

ループで実行された Task オブジェクトでまだ完了していないものの集合を返します。

loop が None の場合、 get_running_loop() は現在のループを取得するのに使われます。

Added in version 3.7.

asyncio.iscoroutine(obj)

obj がコルーチンオブジェクトなら True を返します。

Added in version 3.4.

Task オブジェクト

class asyncio.Task(coro, *, loop=None, name=None, context=None, eager_start=False)

Python コルーチン を実行する Future 類 オブジェクトです。 スレッドセーフではありません。

Task はイベントループのコルーチンを実行するのに使われます。 Future でコルーチンが待機している場合、 Task は自身のコルーチンの実行を一時停止させ、 Future の完了を待ちます。 Future が 完了 したら、 Task が内包しているコルーチンの実行を再開します。

イベントループは協調スケジューリングを使用します。つまり、イベントループは同時に 1 つの Task のみ実行します。 Task が Future の完了を待っているときは、イベントループは他の Task やコールバックを動作させるか、 IO 処理を実行します。

Task を作成するには高レベルの asyncio.create_task() 関数、あるいは低レベルの loop.create_task() 関数や ensure_future() 関数を使用してください。 Task を直接インスタンス化することは推奨されません。

To cancel a running Task use the cancel() method. Calling it will cause the Task to throw a CancelledError exception into the wrapped coroutine. If a coroutine is awaiting on a Future object during cancellation, the Future object will be cancelled.

cancelled() を使うと Task がキャンセルされたかどうかを確認できます。 Task が内包するコルーチンが CancelledError 例外を抑制せず、実際にキャンセルされた場合、このメソッドは True を返します。

asyncio.Task は、Future.set_result() と Future.set_exception() を除いて、Future の API をすべて継承しています。

省略可能なキーワード専用引数 context では、coro を実行するための独自の contextvars.Context を指定できます。 context を指定しない場合、Task は現在のコンテキストをコピーし、後でそのコピーの中でコルーチンを実行します。

省略可能なキーワード専用引数 eager_start では、asyncio.Task の作成時に実行を即座に開始するよう指定できます。 True に設定され、イベントループが実行中であれば、タスクはコルーチンの実行を即座に開始し、コルーチンが初めてブロックするまで実行します。 コルーチンがブロックせずに値を返すか例外を送出すると、タスクはその場で完了し、イベントループへのスケジュールを省略します。

Task は、ラップするコルーチンの戻り値の型を型パラメーターとする ジェネリック型 です。

バージョン 3.7 で変更: contextvars モジュールのサポートを追加。

バージョン 3.8 で変更: name パラメータを追加しました。

バージョン 3.10 で非推奨: loop が指定されておらず、実行中のイベントループもない場合に、非推奨の警告を出すようになりました。

バージョン 3.11 で変更: context パラメータを追加しました。

バージョン 3.12 で変更: eager_start パラメーターが追加されました。

done()

Task が 完了 しているなら True を返します。

Task がラップしているコルーチンが値を返すか、例外を送出するか、または Task がキャンセルされたとき、 Task は 完了 します。

result()

Task の結果を返します。

Task が 完了 している場合、ラップしているコルーチンの結果が返されます (コルーチンが例外を送出された場合、その例外が例外が再送出されます)

Task が キャンセル されている場合、このメソッドは CancelledError 例外を送出します。

Task の結果がまだ利用できない場合、このメソッドは InvalidStateError を送出します。

exception()

Task の例外を返します。

ラップされたコルーチンが例外を送出した場合、その例外が返されます。ラップされたコルーチンが正常終了した場合、このメソッドは None を返します。

Task が キャンセル されている場合、このメソッドは CancelledError 例外を送出します。

Task がまだ 完了 していない場合、このメソッドは InvalidStateError 例外を送出します。

add_done_callback(callback, *, context=None)

Task が 完了 したときに実行されるコールバックを追加します。

このメソッドは低水準のコールバックベースのコードでのみ使うべきです。

詳細については Future.add_done_callback() のドキュメントを参照してください。

remove_done_callback(callback)

コールバックリストから callback を削除します。

このメソッドは低水準のコールバックベースのコードでのみ使うべきです。

詳細については Future.remove_done_callback() のドキュメントを参照してください。

get_stack(*, limit=None)

このタスクのスタックフレームのリストを返します。

コルーチンが完了していない場合、これはサスペンドされた時点でのスタックを返します。コルーチンが正常に処理を完了したか、キャンセルされていた場合は空のリストを返します。コルーチンが例外で終了した場合はトレースバックフレームのリストを返します。

フレームは常に古いものから新しい物へ並んでいます。

サスペンドされているコルーチンの場合スタックフレームが 1 個だけ返されます。

省略可能な limit 引数は返すフレーム数の上限を指定します。 デフォルトでは取得可能なすべてのフレームを返します。 返されるリストの順序は、スタックとトレースバックのどちらを返すかによって異なります。 スタックの場合は最も新しいフレーム群、トレースバックの場合は最も古いフレーム群が返されます (これは traceback モジュールの動作と一致します)。

print_stack(*, limit=None, file=None)

このタスクのスタックまたはトレースバックを出力します。

このメソッドは get_stack() によって取得されるフレームに対し、 traceback モジュールと同じような出力を生成します。

引数 limit は get_stack() にそのまま渡されます。

file 引数は出力先の I/O ストリームです。 デフォルトでは sys.stdout に出力します。

get_coro()

Task がラップしているコルーチンオブジェクトを返します。

注釈

即時実行ですでに完了した Task では None を返します。 即時実行するタスクファクトリー を参照してください。

Added in version 3.8.

バージョン 3.12 で変更: 即時実行するタスクが追加されたため、結果が None になる場合があります。

get_context()

タスクに関連付けられた contextvars.Context オブジェクトを返します。

Added in version 3.12.

get_name()

Task の名前を返します。

Task に対して明示的に名前が設定されていない場合, デフォルトの asyncio Task 実装はタスクをインスタンス化する際にデフォルトの名前を生成します。

Added in version 3.8.

set_name(value)

Task に名前を設定します。

引数 value は文字列に変換可能なオブジェクトであれば何でもかまいません。

デフォルトの Task 実装では、タスクオブジェクトの repr() の出力で名前を確認できます。

Added in version 3.8.

cancel(msg=None)

このタスクに、自身のキャンセルを要求します。

Task がすでに 完了 または キャンセル されている場合は False を、それ以外の場合は True を返します。

このメソッドは、イベントループの次のサイクルで、ラップしているコルーチンに CancelledError を送出するように手配します。

コルーチンは、try ... except CancelledError ... finally ブロックで後片付けをしたり、例外を抑制してキャンセル要求を拒否したりできます。 このため、Future.cancel() と異なり、Task.cancel() は Task がキャンセルされることを保証しません。 ただし、キャンセルを完全に抑制するのは一般的ではなく、強く非推奨とします。 それでもキャンセルを抑制する場合、コルーチンは例外を捕捉するだけでなく Task.uncancel() も呼び出す必要があります。

バージョン 3.9 で変更: msg パラメーターが追加されました。

バージョン 3.11 で変更: msg パラメーターが、キャンセルされたタスクからそのタスクを await する側へ伝播するようになりました。

次の例は、コルーチンがキャンセル要求を捕捉する方法を示しています。

async def cancel_me():
    print('cancel_me(): before sleep')

    try:
        # 1 時間待つ
        await asyncio.sleep(3600)
    except asyncio.CancelledError:
        print('cancel_me(): cancel sleep')
        raise
    finally:
        print('cancel_me(): after sleep')

async def main():
    # "cancel_me" タスクを作成する
    task = asyncio.create_task(cancel_me())

    # 1 秒待つ
    await asyncio.sleep(1)

    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("main(): cancel_me is cancelled now")

asyncio.run(main())

# 期待される出力:
#
#     cancel_me(): before sleep
#     cancel_me(): cancel sleep
#     cancel_me(): after sleep
#     main(): cancel_me is cancelled now

cancelled()

Task が キャンセルされた 場合に True を返します。

cancel() メソッドによりキャンセルがリクエストされ、かつ Task がラップしているコルーチンが内部で送出された CancelledError 例外を伝達したとき、 Task は実際に キャンセル されます。

uncancel()

この Task に対するキャンセル要求の回数を減らします。

残っているキャンセル要求の回数を返します。

キャンセルされたタスクの実行が完了した後は、uncancel() を呼び出しても効果はありません。

Added in version 3.11.

このメソッドは asyncio の内部で使うもので、ユーザーのコードからの使用は想定していません。 特に、Task のキャンセル状態を正常に解除できると、キャンセルの影響をそれぞれの構造化されたブロック内に限定でき、タスクグループ や asyncio.timeout() などの構造化された並行処理を継続できます。 例えば次のようになります。

async def make_request_with_timeout():
    try:
        async with asyncio.timeout(1):
            # タイムアウトの影響を受ける構造化されたブロック:
            await make_request()
            await make_another_request()
    except TimeoutError:
        log("There was a timeout")
    # 外側のコードはタイムアウトの影響を受けない:
    await unrelated_code()

make_request() と make_another_request() を含むブロックはタイムアウトによってキャンセルされる可能性がありますが、その場合も unrelated_code() は実行を続ける必要があります。 これを実現するのが uncancel() です。 TaskGroup コンテキストマネージャーも同様に uncancel() を使います。

何らかの理由でユーザーのコードが CancelledError を捕捉してキャンセルを抑制する場合、キャンセル状態を解除するためにこのメソッドを呼び出す必要があります。

このメソッドによってキャンセル回数がゼロになると、以前の cancel() の呼び出しがタスクへの CancelledError の送出を手配していたかどうかを確認します。 まだ送出されていなければ、内部の _must_cancel フラグをリセットして、その手配を取り消します。

バージョン 3.13 で変更: キャンセル回数がゼロになったら、保留中のキャンセル要求を取り消すようになりました。

cancelling()

この Task に対する保留中のキャンセル要求の回数、つまり cancel() の呼び出し回数から uncancel() の呼び出し回数を引いた値を返します。

この回数がゼロより大きくても、Task がまだ実行中なら cancelled() は False を返します。 uncancel() を呼び出してこの回数を減らし、キャンセル要求がゼロになれば、最終的にタスクがキャンセルされないこともあるためです。

このメソッドは asyncio の内部で使うもので、ユーザーのコードからの使用は想定していません。 詳しくは uncancel() を参照してください。

Added in version 3.11.