Python asyncio.as_completed()とは？終わった順に結果を受け取る方法

asyncio.as_completed()は、終わった順に結果を受け取る関数です。

gather()が全部揃うまで待つのに対し、こちらは1つ終わるたびに処理を始められます。

終わった順に返る

import asyncio

async def work(name, sec):
    await asyncio.sleep　(sec)
    return name

async def main():
    out = []
    for f in asyncio.as_completed([work("遅い", 0.3),
                                   work("中", 0.2),
                                   work("速い", 0.05)]):
        out.append(await f)
    return out

print(asyncio.run(main()))

実行結果：

['速い', '中', '遅い']

渡した順とは逆になりました。早く終わったものから順に返っています。

同じ処理をgather()で書くと引数の順になります。

async def main():
    return await asyncio.gather(work("遅い", 0.3), work("中", 0.2), work("速い", 0.05))

実行結果：

['遅い', '中', '速い']

書き方の形

for f in asyncio.as_completed(tasks):
    result = await f

2段階になっている点が特徴です。

forは普通のfor（async forではない）
中でawait fが必要

返ってくるfは結果そのものではなくawaitableなオブジェクトです。awaitを忘れると結果が取れません。

またどのタスクの結果かは分かりません。返るのは値だけです。

どれの結果か知るには

対応付けが必要なら、結果に識別子を含めて返します。

async def fetch(url, sec):
    await asyncio.sleep(sec)
    return url, f"{url}の中身"

async def main():
    tasks = [fetch("A", 0.3), fetch("B", 0.05)]
    for f in asyncio.as_completed(tasks):
        url, body = await f
        print(url, "→", body)

asyncio.run(main())

実行結果：

B → Bの中身
A → Aの中身

タプルで返すのが簡単な方法です。

Taskを使えば、doneとpendingで管理するasyncio.wait()という選択肢もあります。

使いどころ

結果が届き次第、次の処理に移りたいときに価値があります。

取得できたものから画面に表示する
取得できたものからDBに保存する
進捗を1件ずつ表示する
最初に成功した1件だけ使う

async def main():
    tasks = [fetch(u, 0.1) for u in urls]
    done = 0
    for f in asyncio.as_completed(tasks):
        url, body = await f
        done += 1
        print(f"[{done}/{len(tasks)}] {url} 完了")

asyncio.run(main())

100件の処理で、最後の1件を待たずに99件を処理できます。

逆に「全部揃ってから一括で処理する」ならgather()のほうが素直です。

例外の扱い

async def boom():
    await asyncio.sleep　(0.1)
    raise ValueError("失敗")

async def main():
    for f in asyncio.as_completed([work("A", 0.05), boom()]):
        try:
            print("成功:", await f)
        except Exception as e:
            print("失敗:", type(e).__name__, e)

asyncio.run(main())

実行結果：

成功: A
失敗: ValueError 失敗

例外はawait fの時点で投げられます。

gather()と違い、1つ失敗しても残りは処理が続きます。ループの中でtryを書けば、成功した分を確実に拾えます。

gather()で同じことをするにはreturn_exceptions=Trueが必要です。

gatherとの比較

	`gather()`	`as_completed()`
結果の順序	引数の順	完了順
受け取り方	まとめてリスト	1件ずつ
どれの結果か	分かる	工夫が必要
1件失敗したとき	全体が落ちる	残りは続く
書きやすさ	簡単	やや冗長

迷ったらgather()で十分です。

as_completed()を選ぶのは、「早く届いたものから処理する」ことに意味がある場合だけです。

タイムアウトも設定できる

async def main():
    try:
        for f in asyncio.as_completed([work("遅い", 1)], timeout=0.2):
            print(await f)
    except asyncio.TimeoutError:
        print("タイムアウト")

asyncio.run(main())

実行結果：