Python: Асинхронний Python

Пишіть конкурентні програми з `async def`, `await` та `asyncio`. Охоплює корутини, `asyncio.gather`, завдання, `asyncio.Queue` та асинхронні ітератори.

Короткий вступ до теми та пояснення перед вправами (вправи нижче):

**Навіщо async? Проблема блокуючого I/O** Звичайний сервер більшість часу *чекає* — відповіді бази даних, HTTP-запиту, читання файлу. При синхронному коді потік заблокований і не може нічого робити. Async вирішує це дозволяючи одному потоку обробляти багато задач, що чекають, водночас. Поки одна задача чекає на I/O, цикл подій запускає іншу. **Чим async НЕ є** Async не прискорює CPU-bound роботу. Якщо код витрачає час на обчислення (сортування, обробка зображень), async нічого не дасть — потік зайнятий, а не чекає. Для CPU-роботи використовуйте `multiprocessing`. **Корутини: основна одиниця** Корутина — функція визначена з `async def`. Виклик її НЕ запускає тіло — вона створює об'єкт корутини: ```python import asyncio async def fetch(url: str) -> str: await asyncio.sleep(1) # передає управління циклу подій return f'data from {url}' # Нічого не робить сам по собі: coro = fetch('https://example.com') # об'єкт корутини # Запустити: result = asyncio.run(fetch('https://example.com')) ``` **`await` — призупинення та відновлення** `await expr` призупиняє поточну корутину та передає управління циклу подій. Коли очікуване завершується, цикл відновлює корутину з місця паузи. ```python async def slow_task(name: str, delay: float): print(f'{name}: starting') await asyncio.sleep(delay) print(f'{name}: done after {delay}s') async def main(): # Послідовно — разом 3 секунди: await slow_task('A', 1) await slow_task('B', 2) asyncio.run(main()) ``` **`asyncio.gather` — паралельне виконання задач** `gather` планує кілька корутин паралельно в одному циклі подій: ```python async def main(): # Конкурентно — разом ~2 секунди: results = await asyncio.gather( slow_task('A', 1), slow_task('B', 2), ) asyncio.run(main()) ```

**`asyncio.Task` vs корутина** Корутина виконується тільки коли її awaited. `Task` обгортає корутину і планує її виконання *незалежно* в циклі подій: ```python import asyncio async def background_job(): await asyncio.sleep(5) print('background done') async def main(): task = asyncio.create_task(background_job()) # планується одразу print('doing other work') # виконується зараз await asyncio.sleep(1) await task # чекаємо завершення asyncio.run(main()) ``` **`async for` та `async with`** ```python # async with — async контекстний менеджер async with aiofiles.open('data.txt') as f: content = await f.read() # async for — async ітератор async for line in aiohttp_response.content: process(line) # async генератор: async def lines(filename): async with aiofiles.open(filename) as f: async for line in f: yield line.strip() ``` **Таймаут та скасування** ```python import asyncio async def main(): try: result = await asyncio.wait_for(slow(), timeout=2.0) except asyncio.TimeoutError: print('timed out') # Python 3.11+: async with asyncio.timeout(2.0): result = await slow() ``` **Реальний патерн: конкурентні HTTP-запити** ```python import asyncio import aiohttp async def fetch_url(session, url): async with session.get(url) as response: return await response.text() async def fetch_all(urls): async with aiohttp.ClientSession() as session: tasks = [fetch_url(session, url) for url in urls] return await asyncio.gather(*tasks) results = asyncio.run(fetch_all(['https://example.com', 'https://example.org'])) # Обидва завантажуються конкурентно — час = найповільніший, не сума ```

**Поширені помилки в async** **1. Блокування циклу подій** Будь-який звичайний блокуючий виклик заморожує весь цикл подій: ```python import asyncio, time async def bad(): time.sleep(5) # БЛОКУЄ — заморожує весь цикл подій return 'done' async def good(): await asyncio.sleep(5) # передає управління return 'done' ``` Для CPU-роботи або неминучих блокуючих викликів використовуйте `run_in_executor`: ```python async def main(): loop = asyncio.get_event_loop() with ThreadPoolExecutor() as pool: result = await loop.run_in_executor(pool, blocking_call, arg) ``` **2. Забутий `await`** ```python async def main(): result = fetch() # БАГ: створює об'єкт, ніколи не виконується result = await fetch() # правильно ``` **3. Змішування sync та async** ```python def sync_func(): await something() # SyntaxError: 'await' поза async функцією ``` Якщо потрібно викликати async з sync-контексту — використовуйте `asyncio.run()`. Не викликайте `asyncio.run()` всередині вже запущеного циклу подій. **asyncio vs потоки vs multiprocessing** | Навантаження | Рішення | Чому | |---|---|---| | I/O-bound (мережа, файли, БД) | `asyncio` | Один потік, без накладних | | I/O-bound + блокуючі бібліотеки | `threading` | | | CPU-bound | `multiprocessing` | Обходить GIL | | Змішаний I/O + CPU | `asyncio` + `run_in_executor` | | **Швидкий довідник** | Концепція | Код | |---|---| | Визначити корутину | `async def func(): ...` | | Await корутину | `result = await coro()` | | Запустити з sync | `asyncio.run(main())` | | Конкурентно | `asyncio.gather(c1, c2)` | | Незалежне планування | `asyncio.create_task(coro())` | | Таймаут | `asyncio.wait_for(coro(), timeout=N)` | | Async контекст | `async with expr as x:` | | Async ітератор | `async for item in expr:` |

Вправи:

Напишіть корутину `async def greet(name: str)`, що виводить `'Привіт, <name>!'`, очікує 0.1 секунди через `asyncio.sleep`, потім виводить `'До побачення, <name>!'`. Запустіть її через `asyncio.run(greet('Аліса'))`.

import asyncio

async def greet(name: str) -> None:
    pass


asyncio.run(greet('Аліса'))
# Привіт, Аліса!
# До побачення, Аліса!

Рішення

import asyncio

async def greet(name: str) -> None:
    print(f'Привіт, {name}!')
    await asyncio.sleep(0.1)
    print(f'До побачення, {name}!')


asyncio.run(greet('Аліса'))
# Привіт, Аліса!
# До побачення, Аліса!

Напишіть корутину `fetch(url: str, delay: float) -> str`, що симулює HTTP-запит: чекає `delay` секунд і повертає `f'Response from {url}'`. Використайте `asyncio.gather` для запуску трьох fetch конкурентно: `fetch('A', 0.3)`, `fetch('B', 0.1)`, `fetch('C', 0.2)`. Виведіть всі результати. Загальний час — близько 0.3с, а не 0.6с.

import asyncio

async def fetch(url: str, delay: float) -> str:
    pass

async def main() -> None:
    results = await asyncio.gather(
        fetch('A', 0.3),
        fetch('B', 0.1),
        fetch('C', 0.2),
    )
    for r in results:
        print(r)


asyncio.run(main())

Рішення

import asyncio

async def fetch(url: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f'Response from {url}'

async def main() -> None:
    results = await asyncio.gather(
        fetch('A', 0.3),
        fetch('B', 0.1),
        fetch('C', 0.2),
    )
    for r in results:
        print(r)


asyncio.run(main())
# Response from A
# Response from B
# Response from C

Напишіть корутину `main()`, що створює два фонові завдання через `asyncio.create_task`: одне відраховує від 3 до 1 (виводить кожне число з затримкою 0.1с), інше виводить `'Працюємо...'` тричі із затримкою 0.15с. Дочекайтесь обох завдань наприкінці. Зверніть увагу, що виводи переплітаються.

import asyncio

async def countdown() -> None:
    for i in range(3, 0, -1):
        print(f'Зворотній відлік: {i}')
        await asyncio.sleep(0.1)

async def worker() -> None:
    for _ in range(3):
        print('Працюємо...')
        await asyncio.sleep(0.15)

async def main() -> None:
    # створіть завдання та дочекайтесь їх
    pass


asyncio.run(main())

Рішення

import asyncio

async def countdown() -> None:
    for i in range(3, 0, -1):
        print(f'Зворотній відлік: {i}')
        await asyncio.sleep(0.1)

async def worker() -> None:
    for _ in range(3):
        print('Працюємо...')
        await asyncio.sleep(0.15)

async def main() -> None:
    task1 = asyncio.create_task(countdown())
    task2 = asyncio.create_task(worker())
    await task1
    await task2


asyncio.run(main())

Реалізуйте патерн виробник-споживач з `asyncio.Queue`. Корутина `producer` кладе числа 1–5 в чергу (затримка 0.05с між числами), потім кладе `None` як сигнал завершення. Корутина `consumer` читає елементи з черги до появи `None`, виводячи `'Consuming: <n>'` для кожного числа.

import asyncio

async def producer(queue: asyncio.Queue) -> None:
    pass

async def consumer(queue: asyncio.Queue) -> None:
    pass

async def main() -> None:
    queue: asyncio.Queue = asyncio.Queue()
    await asyncio.gather(producer(queue), consumer(queue))


asyncio.run(main())

Рішення

import asyncio

async def producer(queue: asyncio.Queue) -> None:
    for i in range(1, 6):
        await queue.put(i)
        await asyncio.sleep(0.05)
    await queue.put(None)  # сигнал завершення

async def consumer(queue: asyncio.Queue) -> None:
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f'Consuming: {item}')

async def main() -> None:
    queue: asyncio.Queue = asyncio.Queue()
    await asyncio.gather(producer(queue), consumer(queue))


asyncio.run(main())

Напишіть корутину `slow_op(seconds: float) -> str`, що чекає `seconds` секунд та повертає `'done'`. Використайте `asyncio.wait_for(slow_op(5), timeout=0.2)` для спроби операції з таймаутом 0.2 секунди. Перехопіть `asyncio.TimeoutError` та виведіть `'Operation timed out!'`.

import asyncio

async def slow_op(seconds: float) -> str:
    await asyncio.sleep(seconds)
    return 'done'

async def main() -> None:
    try:
        result = await asyncio.wait_for(slow_op(5), timeout=0.2)
        print(result)
    except asyncio.TimeoutError:
        pass  # виведіть повідомлення тут


asyncio.run(main())
# Operation timed out!

Рішення

import asyncio

async def slow_op(seconds: float) -> str:
    await asyncio.sleep(seconds)
    return 'done'

async def main() -> None:
    try:
        result = await asyncio.wait_for(slow_op(5), timeout=0.2)
        print(result)
    except asyncio.TimeoutError:
        print('Operation timed out!')


asyncio.run(main())
# Operation timed out!

Напишіть асинхронний генератор `async_range(start: int, stop: int, delay: float)`, що видає цілі числа від `start` до `stop-1`, роблячи паузу `delay` секунд між кожним значенням. Використайте `async for` для споживання та виводу значень.

import asyncio

async def async_range(start: int, stop: int, delay: float):
    pass

async def main() -> None:
    async for value in async_range(0, 4, 0.05):
        print(value)


asyncio.run(main())
# 0
# 1
# 2
# 3

Рішення

import asyncio

async def async_range(start: int, stop: int, delay: float):
    for i in range(start, stop):
        yield i
        await asyncio.sleep(delay)

async def main() -> None:
    async for value in async_range(0, 4, 0.05):
        print(value)


asyncio.run(main())

Створіть клас `AsyncTimer`, що працює як асинхронний контекстний менеджер. `__aenter__` фіксує час початку та виводить `'Timer started'`. `__aexit__` обчислює час виконання та виводить `f'Elapsed: {elapsed:.2f}s'`. Використайте з `async with AsyncTimer():`.

import asyncio
import time

class AsyncTimer:
    async def __aenter__(self):
        pass

    async def __aexit__(self, *args):
        pass

async def main() -> None:
    async with AsyncTimer():
        await asyncio.sleep(0.1)


asyncio.run(main())
# Timer started
# Elapsed: 0.10s

Рішення

import asyncio
import time

class AsyncTimer:
    async def __aenter__(self):
        self._start = time.perf_counter()
        print('Timer started')
        return self

    async def __aexit__(self, *args):
        elapsed = time.perf_counter() - self._start
        print(f'Elapsed: {elapsed:.2f}s')

async def main() -> None:
    async with AsyncTimer():
        await asyncio.sleep(0.1)


asyncio.run(main())

У вас є 8 завдань, але одночасно має виконуватися не більше 3. Використайте `asyncio.Semaphore(3)` для обмеження конкурентності. Кожне завдання виводить `'Task N started'`, чекає 0.1с, потім `'Task N done'`. Запустіть всі 8 завдань через `asyncio.gather`.

import asyncio

async def task(n: int, sem: asyncio.Semaphore) -> None:
    async with sem:
        print(f'Task {n} started')
        await asyncio.sleep(0.1)
        print(f'Task {n} done')

async def main() -> None:
    sem = asyncio.Semaphore(3)
    # створіть і запустіть 8 завдань (n від 1 до 8)
    pass


asyncio.run(main())

Рішення

import asyncio

async def task(n: int, sem: asyncio.Semaphore) -> None:
    async with sem:
        print(f'Task {n} started')
        await asyncio.sleep(0.1)
        print(f'Task {n} done')

async def main() -> None:
    sem = asyncio.Semaphore(3)
    await asyncio.gather(*(task(n, sem) for n in range(1, 9)))


asyncio.run(main())

Функція `blocking_sum(n)` обчислює суму 1..n через цикл (симулює повільну роботу). Оберніть її в асинхронну `async_sum(n)`, використовуючи `loop.run_in_executor(None, blocking_sum, n)`, щоб не блокувати цикл подій. Запустіть два виклики `async_sum` конкурентно через `asyncio.gather`.

import asyncio

def blocking_sum(n: int) -> int:
    return sum(range(n + 1))

async def async_sum(n: int) -> int:
    loop = asyncio.get_running_loop()
    # запустіть blocking_sum у пулі потоків
    pass

async def main() -> None:
    results = await asyncio.gather(async_sum(1000), async_sum(2000))
    print(results)  # [500500, 2001000]


asyncio.run(main())

Рішення

import asyncio

def blocking_sum(n: int) -> int:
    return sum(range(n + 1))

async def async_sum(n: int) -> int:
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, blocking_sum, n)

async def main() -> None:
    results = await asyncio.gather(async_sum(1000), async_sum(2000))
    print(results)  # [500500, 2001000]


asyncio.run(main())

Використайте `asyncio.Event` для координації двох корутин. Корутина `setter` чекає 0.15с, виводить `'Setting event'`, потім встановлює подію. Корутина `waiter` виводить `'Waiting for event...'`, чекає через `await event.wait()`, потім виводить `'Event received!'`. Запустіть обидві через `asyncio.gather`.

import asyncio

async def setter(event: asyncio.Event) -> None:
    await asyncio.sleep(0.15)
    print('Setting event')
    event.set()

async def waiter(event: asyncio.Event) -> None:
    pass  # зачекайте подію, потім виведіть 'Event received!'

async def main() -> None:
    event = asyncio.Event()
    await asyncio.gather(setter(event), waiter(event))


asyncio.run(main())

Рішення

import asyncio

async def setter(event: asyncio.Event) -> None:
    await asyncio.sleep(0.15)
    print('Setting event')
    event.set()

async def waiter(event: asyncio.Event) -> None:
    print('Waiting for event...')
    await event.wait()
    print('Event received!')

async def main() -> None:
    event = asyncio.Event()
    await asyncio.gather(setter(event), waiter(event))


asyncio.run(main())

HOME

COURSES

TYPING APP

CODE PLAYGROUND

ВПРАВИ

REGEX TESTER

НАЛАШТУВАННЯ

ПРО САЙТ

Асинхронний Python

Цикл подій, корутини, await та gather детально

Tasks, async for/with, таймаут, конкурентні HTTP-запити

Поширені помилки, asyncio vs потоки vs multiprocessing, довідник

Перша корутина з asyncio.run

Конкурентні завдання з asyncio.gather

Фонові завдання з create_task

Виробник-споживач з asyncio.Queue

Таймаут з asyncio.wait_for

Асинхронний генератор

Асинхронний контекстний менеджер

Обмеження конкурентності з asyncio.Semaphore

Виконання блокуючого коду через run_in_executor

Координація з asyncio.Event

Навчайтеся, читаючи, запускаючи код і розв'язуючи задачі.

Навчальні треки

Запуск коду

Перевірка патернів

Тренування друку

Як працює цей workspace

Одна IDE, три практичні режими

Працюйте поруч у split tabs

Тренажер друку коду

HOME

COURSES

TYPING APP

CODE PLAYGROUND

Ліміти файлів проєкту

ВПРАВИ

REGEX TESTER

НАЛАШТУВАННЯ

ПРО САЙТ

Навчальні треки

Запуск коду

Перевірка патернів

Тренування друку

Як працює цей workspace

Одна IDE, три практичні режими

Працюйте поруч у split tabs