モダンPython上級 #4 非同期の深さ — イベントループ、gather/wait、async generator

def run_loop(coros):
    queue = list(coros)
    while queue:
        coro = queue.pop(0)
        try:
            # コルーチンを次の await まで進める
            future = coro.send(None)
        except StopIteration as e:
            continue   # コルーチン終了
        # future が終わったらコルーチンをキューに戻す
        future.add_done_callback(lambda _: queue.append(coro))

async def slow():
    total = 0
    for i in range(10_000_000):    # CPU バウンド、await なし
        total += i
    return total

async def main():
    await asyncio.gather(slow(), slow(), slow())   # 3 つを同時に?

async def main():
    loop = asyncio.get_running_loop()
    fut = loop.create_future()

    # 1 秒後に結果を埋める
    loop.call_later(1, fut.set_result, "done")

    result = await fut
    print(result)   # done

async def main():
    task = asyncio.create_task(work())   # Task の生成
    print(isinstance(task, asyncio.Future))   # True

task = asyncio.create_task(work())

await asyncio.sleep(0.1)
print(task.done())     # False / True

if task.done():
    if task.exception():
        print("失敗:", task.exception())
    else:
        print("成功:", task.result())

task.cancel()           # キャンセル試行

async def main():
    asyncio.create_task(background_work())   # 結果を無視
    # ...

async def main():
    background_tasks = set()

    task = asyncio.create_task(background_work())
    background_tasks.add(task)
    task.add_done_callback(background_tasks.discard)

    # ...

results = await asyncio.gather(
    fetch("a"), fetch("b"), fetch("c"),
    return_exceptions=False,
)
# results: [a の結果, b の結果, c の結果]  ← 呼び出し順

results = await asyncio.gather(
    fetch("a"),
    fetch_might_fail("b"),
    return_exceptions=True,
)
for r in results:
    if isinstance(r, Exception):
        print("失敗:", r)
    else:
        print("成功:", r)

done, pending = await asyncio.wait(
    [task1, task2, task3],
    return_when=asyncio.FIRST_COMPLETED,
)

async def main():
    coros = [fetch(u) for u in urls]
    for coro in asyncio.as_completed(coros):
        result = await coro
        print("受信:", result)

import asyncio
import requests   # 同期ライブラリ

async def fetch(url):
    response = await asyncio.to_thread(requests.get, url)
    return response.text

from concurrent.futures import ProcessPoolExecutor

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_heavy_function, arg)

async def stream_chunks(url):
    async with httpx.AsyncClient() as client:
        async with client.stream("GET", url) as resp:
            async for chunk in resp.aiter_bytes():
                yield chunk

async def main():
    async for chunk in stream_chunks("https://example.com"):
        print(len(chunk))

class Counter:
    def __init__(self, limit):
        self.limit = limit
        self.n = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.n >= self.limit:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)
        self.n += 1
        return self.n

async def main():
    async for i in Counter(3):
        print(i)
# 0.1 秒間隔で 1, 2, 3

from contextlib import asynccontextmanager

@asynccontextmanager
async def open_db():
    conn = await connect_async()
    try:
        yield conn
    finally:
        await conn.close()

async def main():
    async with open_db() as conn:
        await conn.execute(...)

sem = asyncio.Semaphore(10)

async def fetch_one(url):
    async with sem:
        return await fetch(url)

async def fetch_all(urls):
    return await asyncio.gather(*[fetch_one(u) for u in urls])

queue = asyncio.Queue(maxsize=10)

async def producer():
    for i in range(100):
        await queue.put(i)
    await queue.put(None)    # 終了シグナル

async def consumer():
    while True:
        item = await queue.get()
        if item is None:
            break
        process(item)
        queue.task_done()

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer())
        tg.create_task(consumer())

async def main():
    tasks = [
        asyncio.create_task(fetch_a()),
        asyncio.create_task(fetch_b()),
    ]
    done, pending = await asyncio.wait(
        tasks, return_when=asyncio.FIRST_COMPLETED
    )
    for t in pending:
        t.cancel()
    return done.pop().result()

async def main():
    try:
        async with asyncio.timeout(5):
            data = await slow_op()
    except TimeoutError:
        print("5 秒超過")

async def work():
    try:
        await long_operation()
    except asyncio.CancelledError:
        await cleanup()    # 後片付けは非同期でも可能
        raise              # ← 必ず再送出!

async def critical():
    return await asyncio.shield(important_work())

PYTHONASYNCIODEBUG=1 python app.py