Coroutines and tasks

この節では、コルーチンと 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() です。

Creating tasks

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

---

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

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

If name is not None, it is set as the name of the task using Task.set_name().

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

その 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)

Note that this approach never awaits the tasks, so if a task fails, its exception is never retrieved and asyncio logs a "Task exception was never retrieved" message when the task is garbage collected. To avoid this, use asyncio.TaskGroup which keeps a strong reference to each task, awaits them and propagates their exceptions:

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 パラメータを追加しました。

Task cancellation

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

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

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

Task groups

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

class asyncio.TaskGroup

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

Added in version 3.11.

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

Create a task in this task group. The signature matches that of asyncio.create_task(). If the task group is inactive (e.g. not yet entered, already finished, or in the process of shutting down), we will close the given coro and raise RuntimeError.

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

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

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())

予想される出力:

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 を送出するようになりました。

Running tasks concurrently

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 のキャンセルを引き起こすのを避けるためです。

注釈

A new alternative to create and run tasks concurrently and wait for their completion is asyncio.TaskGroup. TaskGroup provides stronger safety guarantees than gather for scheduling a nesting of subtasks: if a task (or a subtask, a task scheduled by a task) raises an exception, TaskGroup will, while gather will not, cancel the remaining scheduled tasks.

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

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 のようなオブジェクトというわけではなく、かつ実行中のイベントループがない場合に、非推奨の警告を出すようになりました。

Eager task factory

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.

Shielding from cancellation

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)

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

Prefer using asyncio.timeout() or asyncio.timeout_at() rather than instantiating Timeout directly.

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 が完了するまで待ちます。

If a timeout occurs, it cancels aw and raises TimeoutError.

To prevent aw from being cancelled, wrap it in 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 で変更: Implemented using asyncio.timeout(), a coroutine passed as aw is no longer wrapped in a Task when timeout is positive.

Waiting primitives

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 をキャンセルしません。

If wait() is cancelled, the futures in aws are not cancelled and continue to run.

バージョン 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() does not cancel the tasks running the supplied awaitables: if a timeout occurs or the iteration is cancelled, the remaining tasks continue to run.

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

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

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

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

Running in threads

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.

Scheduling from other threads

asyncio.run_coroutine_threadsafe(coro, loop)

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

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

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

# Create a coroutine
coro = asyncio.sleep(1, result=3)

# Submit the coroutine to a given loop
future = asyncio.run_coroutine_threadsafe(coro, loop)

# Wait for the result with an optional timeout argument
assert future.result(timeout) == 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 object

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 を直接インスタンス化することは推奨されません。

実行中のタスクをキャンセルするためには、cancel() メソッドを使用します。このメソッドを呼ぶと、タスクはそれを内包するコルーチンに対して CancelledError 例外を送出します。キャンセルの際にコルーチンが Future オブジェクトを待っていた場合、その Future オブジェクトはキャンセルされます。

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 に設定され、イベントループが実行中であれば、タスクはコルーチンの実行を即座に開始し、コルーチンが初めてブロックするまで実行します。 コルーチンがブロックせずに値を返すか例外を送出すると、タスクはその場で完了し、イベントループへのスケジュールを省略します。

Tasks are generic over the return type of their wrapped coroutines.

バージョン 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 のキャンセル状態を正常に解除できると、キャンセルの影響をそれぞれの構造化されたブロック内に限定でき、Task groups や 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.