asyncio на практике: задачи, таймауты, семафоры

import asyncio

async def fetch(name, delay, fail=False):
    await asyncio.sleep(delay)
    if fail:
        raise ConnectionError(f"{name}: сервер недоступен")
    return f"{name}: ok"

async def main():
    async with asyncio.TaskGroup() as tg:                   # все задачи живут внутри блока
        users = tg.create_task(fetch("users", 0.1))
        orders = tg.create_task(fetch("orders", 0.2))
    print(users.result(), "|", orders.result())             # после блока все завершены

    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch("users", 0.1, fail=True))
            slow = tg.create_task(fetch("reports", 1.0))    # будет отменена
    except* ConnectionError as eg:
        print("ошибки:", [str(e) for e in eg.exceptions])
    print("reports отменена:", slow.cancelled())

asyncio.run(main())

import asyncio

async def ok(x):
    await asyncio.sleep(0.01)
    return x

async def boom():
    await asyncio.sleep(0.01)
    raise ValueError("сломалось")

async def main():
    try:
        await asyncio.gather(ok(1), boom(), ok(3))        # первое исключение пробрасывается
    except ValueError as e:
        print("gather бросил:", e)

    results = await asyncio.gather(ok(1), boom(), ok(3), return_exceptions=True)
    print(results)                                        # исключения — значениями
    good = [r for r in results if not isinstance(r, Exception)]
    print("успешные:", good)

asyncio.run(main())

import asyncio

async def long_job():
    try:
        print("  работа началась")
        await asyncio.sleep(10)
        return "результат"
    except asyncio.CancelledError:
        print("  отменили — освобождаю ресурсы")
        raise                                     # CancelledError нужно пробрасывать!

async def main():
    task = asyncio.create_task(long_job())
    await asyncio.sleep(0.05)
    task.cancel()                                 # попросить задачу остановиться
    try:
        await task
    except asyncio.CancelledError:
        print("задача отменена:", task.cancelled())

    try:
        async with asyncio.timeout(0.1):          # 3.11+: таймаут на блок кода
            await long_job()
    except TimeoutError:
        print("таймаут через asyncio.timeout")

    try:
        await asyncio.wait_for(long_job(), timeout=0.1)   # таймаут на одну корутину
    except TimeoutError:
        print("таймаут через wait_for")

asyncio.run(main())

import asyncio, random

async def fetch(i):
    delay = random.Random(i).uniform(0.05, 0.3)
    await asyncio.sleep(delay)
    return i, round(delay, 2)

async def main():
    for coro in asyncio.as_completed([fetch(i) for i in range(5)]):
        i, delay = await coro
        print(f"  готов запрос {i} за {delay} с")

asyncio.run(main())

import asyncio, time

async def call_api(i, sem, log):
    async with sem:                         # не больше 3 одновременно
        log.append(f"+{i}")
        await asyncio.sleep(0.1)
        log.append(f"-{i}")
        return i

async def main():
    sem = asyncio.Semaphore(3)
    log = []
    start = time.perf_counter()
    results = await asyncio.gather(*(call_api(i, sem, log) for i in range(9)))
    print(results, f"{time.perf_counter() - start:.1f} с")   # 9 задач по 3 → ≈0.3 с
    print(" ".join(log))

asyncio.run(main())

import asyncio

async def worker(name, queue, results):
    while True:
        url = await queue.get()
        try:
            await asyncio.sleep(0.05)           # «скачиваем»
            results.append((name, url))
        finally:
            queue.task_done()

async def main():
    queue = asyncio.Queue()
    for i in range(10):
        queue.put_nowait(f"/page/{i}")
    results = []
    workers = [asyncio.create_task(worker(f"w{n}", queue, results)) for n in range(3)]
    await queue.join()                          # дождаться обработки всех элементов
    for w in workers:
        w.cancel()                              # остановить бесконечные воркеры
    await asyncio.gather(*workers, return_exceptions=True)
    print(len(results), "страниц;", sorted({name for name, _ in results}))

asyncio.run(main())

import asyncio

balance = 100

async def withdraw(amount, lock=None):
    global balance
    async def critical():
        global balance
        current = balance
        await asyncio.sleep(0.01)          # переключение посреди «прочитать–изменить–записать»
        if current >= amount:
            balance = current - amount
            return True
        return False
    if lock:
        async with lock:
            return await critical()
    return await critical()

async def main():
    global balance
    print(await asyncio.gather(withdraw(80), withdraw(80)), "баланс", balance)   # обе «успешны»!
    balance = 100
    lock = asyncio.Lock()
    print(await asyncio.gather(withdraw(80, lock), withdraw(80, lock)), "баланс", balance)

asyncio.run(main())