import asyncio
import time
from typing import List
 
# ❌ Синхронная версия
def sync_download(file_id: int) -> str:
    """Блокирующая загрузка файла."""
 
    print(f"[{file_id}] Начало загрузки")
    time.sleep(2)  # ⚠️ Блокирует весь процесс!
    print(f"[{file_id}] Загрузка завершена")
    return f"Данные {file_id}"
 
# ✅ Асинхронная версия
async def async_download(file_id: int) -> str:
    """Асинхронная загрузка файла."""
 
    print(f"[{file_id}] Начало загрузки")
    await asyncio.sleep(2)  # ✅ Освобождает event loop
    print(f"[{file_id}] Загрузка завершена")
    return f"Данные {file_id}"
 
# Сравнение производительности
async def compare() -> None:
    """Демонстрация разницы между синхронным и асинхронным подходами."""
 
    # Синхронно: загружаем 3 файла по очереди
    start = time.time()
    results_sync: List[str] = []
    for i in range(3):
        results_sync.append(sync_download(i))
    print(f"Синхронно: {time.time() - start:.1f} сек")  # ~6 секунд
    print(f"Результаты: {results_sync}\n")
 
    # Асинхронно: загружаем 3 файла одновременно
    start = time.time()
    results_async: List[str] = await asyncio.gather(
        async_download(0),
        async_download(1),
        async_download(2)
    )
    print(f"Асинхронно: {time.time() - start:.1f} сек")  # ~2 секунды!
    print(f"Результаты: {results_async}")
 
if __name__ == "__main__":
    asyncio.run(compare())

from typing import Coroutine
 
async def my_coroutine() -> str:
    """Корутина — функция, которая может приостанавливаться на await."""
    return "result"
 
# ❌ ОШИБКА: вызов без await возвращает coroutine объект
coro: Coroutine = my_coroutine()  # <coroutine object my_coroutine>
 
# ✅ ПРАВИЛЬНО: await получает результат
result: str = await my_coroutine()  # "result"

import asyncio
 
async def fetch_data() -> dict:
    # Приостанавливаем выполнение текущей корутины
    # Event loop переключается на другие задачи
    await asyncio.sleep(1)
    return {"status": "ok"}
 
# Пока ждём sleep(1), event loop выполняет другие корутины

async def main() -> None:
    """Точка входа в асинхронное приложение."""
 
    result = await my_coroutine()
    print(result)
 
# ✅ Вызывается ТОЛЬКО в __main__
if __name__ == "__main__":
    asyncio.run(main())
 
# ❌ НИКОГДА не вызывайте asyncio.run() внутри async функции!
async def wrong() -> None:
    asyncio.run(main())  # RuntimeError: asyncio.run() cannot be called from a running event loop

from typing import Coroutine
 
# Обычная функция — выполняется немедленно
def regular_function() -> str:
    return "result"
 
result = regular_function()  # Сразу получаем "result"
print(result)  # "result"
 
# Корутина — возвращает coroutine объект
async def coroutine_function() -> str:
    return "result"
 
coro: Coroutine = coroutine_function()  # Возвращает <coroutine object>
print(coro)  # <coroutine object coroutine_function at 0x...>
 
# Чтобы получить результат, нужен await
result = await coroutine_function()  # "result"

# Генератор — производит значения через yield
def generator():
    yield 1
    yield 2
    yield 3
 
for value in generator():
    print(value)  # 1, 2, 3
 
# Корутина — приостанавливается на await
async def coroutine():
    await asyncio.sleep(1)
    return "done"
 
# Корутины нельзя итерировать, их нужно await
result = await coroutine()  # "done"

import asyncio
import inspect
 
async def my_coroutine() -> str:
    await asyncio.sleep(1)
    return "result"
 
async def demo():
    # Создана, но не запущена
    coro = my_coroutine()
    print(f"State: {inspect.getcoroutinestate(coro)}")  # CORO_CREATED
 
    # Запущена, но приостановлена на await
    task = asyncio.create_task(coro)
    await asyncio.sleep(0.1)
    print(f"State: {inspect.getcoroutinestate(coro)}")  # CORO_SUSPENDED
 
    # Завершена
    result = await task
    print(f"State: {inspect.getcoroutinestate(coro)}")  # CORO_CLOSED
    print(f"Result: {result}")

async def multi_suspend() -> None:
    """Корутина с множественными приостановками."""
 
    print("Начало работы")
 
    await asyncio.sleep(1)  # Приостановка 1
    print("После 1 секунды")
 
    await asyncio.sleep(1)  # Приостановка 2
    print("После 2 секунд")
 
    result = await fetch_data()  # Приостановка 3
    print(f"Получены данные: {result}")
 
    await asyncio.sleep(1)  # Приостановка 4
    print("Завершение работы")

import asyncio
 
async def example() -> None:
    """Пример работы корутины."""
    print("1. Начало выполнения")
 
    # Здесь корутина приостанавливается
    # Управление возвращается в event loop
    await asyncio.sleep(1)
 
    # Через 1 секунду event loop возобновляет выполнение
    print("2. Продолжение после паузы")
 
# Что происходит:
# 1. Корутина начинает выполняться
# 2. Достигает await — приостанавливается
# 3. Event loop переключается на другие задачи
# 4. Через 1 сек event loop возвращается к корутине
# 5. Корутина продолжает выполнение

# ❌ НЕПРАВИЛЬНО — корутина не выполнится
async def wrong():
    my_coroutine()  # Warning: coroutine was never awaited
 
# ✅ ПРАВИЛЬНО
async def correct():
    await my_coroutine()  # Корутина выполнится
 
# ✅ Альтернатива — create_task для фонового выполнения
async def also_correct():
    task = asyncio.create_task(my_coroutine())
    # Задача выполняется в фоне
    await task  # Ждём результат

# ❌ НЕПРАВИЛЬНО — нельзя await в обычной функции
def sync_function():
    result = await async_function()  # SyntaxError!
 
# ✅ ПРАВИЛЬНО — await только внутри async def
async def async_function():
    result = await another_async_function()
    return result
 
# Для запуска из синхронного кода используйте asyncio.run()
def main():
    result = asyncio.run(async_function())

# ❌ НЕПРАВИЛЬНО — старый способ (Python 3.4)
@asyncio.coroutine
def old_style():
    yield from asyncio.sleep(1)
 
# ✅ ПРАВИЛЬНО — современный способ (Python 3.5+)
async def new_style():
    await asyncio.sleep(1)

import asyncio
from typing import List, Any
 
async def cook_pasta() -> str:
    """Асинхронное приготовление пасты (15 секунд)."""
 
    print("Ставлю воду на плиту")
    await asyncio.sleep(5)  # ← Приостановка на 5 сек
    print("Вода закипела!")
    await asyncio.sleep(10)  # ← Ещё одна приостановка
    print("Паста готова!")
    return "Паста"
 
async def make_salad() -> str:
    """Асинхронное приготовление салата (3 секунды)."""
 
    print("Режу овощи")
    await asyncio.sleep(3)
    print("Салат готов!")
    return "Салат"
 
# Готовим параллельно
async def cook_dinner() -> List[str]:
    """
    Параллельное приготовление блюд.
 
    Общее время = max(15, 3) = 15 секунд (не 15 + 3 = 18!)
    """
 
    results: List[str] = await asyncio.gather(
        cook_pasta(),   # 15 секунд
        make_salad()    # 3 секунды
    )
    return results
 
if __name__ == "__main__":
    dishes = asyncio.run(cook_dinner())
    print(f"Готовые блюда: {dishes}")

import asyncio
from typing import List
 
async def task(task_id: int) -> int:
    """Асинхронная задача, возвращающая результат через 1 секунду."""
 
    await asyncio.sleep(1)
    return task_id * 10
 
# 1️⃣ await — последовательное выполнение
async def sequential_execution() -> None:
    """Выполняет задачи последовательно (медленно)."""
 
    result1 = await task(1)  # Ждём 1 сек
    result2 = await task(2)  # Ждём ещё 1 сек
    result3 = await task(3)  # Ждём ещё 1 сек
    print(f"Последовательно (3 сек): {[result1, result2, result3]}")
 
# 2️⃣ create_task — параллельное выполнение с контролем
async def parallel_with_tasks() -> None:
    """Запускает задачи параллельно с возможностью отмены."""
 
    # Создаём задачи (начинают выполняться сразу!)
    task1 = asyncio.create_task(task(1))
    task2 = asyncio.create_task(task(2))
    task3 = asyncio.create_task(task(3))
 
    # Можем делать что-то ещё...
    print("Задачи выполняются в фоне")
    await asyncio.sleep(0.5)  # Имитация другой работы
 
    # Ждём результаты (общее время ~1 сек, не 3!)
    result1 = await task1
    result2 = await task2
    result3 = await task3
    print(f"С create_task (~1 сек): {[result1, result2, result3]}")
 
# 3️⃣ gather — запуск нескольких задач одновременно
async def parallel_with_gather() -> None:
    """Самый простой способ запустить несколько задач параллельно."""
 
    results: List[int] = await asyncio.gather(
        task(1),
        task(2),
        task(3)
    )
    print(f"С gather (~1 сек): {results}")  # [10, 20, 30]
 
if __name__ == "__main__":
    asyncio.run(sequential_execution())      # ~3 секунды
    asyncio.run(parallel_with_tasks())       # ~1 секунда
    asyncio.run(parallel_with_gather())      # ~1 секунда

import asyncio
 
async def task(name, delay):
    print(f"[{name}] Начало")
    await asyncio.sleep(delay)
    print(f"[{name}] Конец")
 
async def main():
    await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 3)
    )
 
asyncio.run(main())

import asyncio
import time
import requests  # Блокирующая библиотека
 
def blocking_api_call(url: str) -> dict:
    """Синхронная функция, которая блокирует thread."""
 
    response = requests.get(url)  # Блокирующий I/O
    time.sleep(1)  # Блокирующая операция
    return response.json()
 
async def fetch_data_from_blocking_lib():
    """Правильное использование блокирующей библиотеки в async коде."""
 
    loop = asyncio.get_running_loop()
 
    # ✅ Выполняем в отдельном потоке, не блокируя event loop
    result = await loop.run_in_executor(
        None,  # None = default ThreadPoolExecutor
        blocking_api_call,
        "https://api.example.com/data"
    )
 
    return result
 
# Альтернатива с functools.partial для передачи аргументов
from functools import partial
 
async def fetch_multiple_blocking():
    """Параллельное выполнение нескольких блокирующих операций."""
 
    loop = asyncio.get_running_loop()
 
    urls = [
        "https://api.example.com/1",
        "https://api.example.com/2",
        "https://api.example.com/3"
    ]
 
    # Параллельно вызываем блокирующие функции в threads
    tasks = [
        loop.run_in_executor(None, partial(blocking_api_call, url))
        for url in urls
    ]
 
    results = await asyncio.gather(*tasks)
    return results

import asyncio
 
def callback_function():
    """Обычная синхронная функция."""
 
    print("Callback выполнен!")
 
async def schedule_callbacks():
    """Примеры планирования callbacks через loop."""
 
    loop = asyncio.get_running_loop()
 
    # Выполнить callback как можно скорее
    loop.call_soon(callback_function)
 
    # Выполнить callback через 5 секунд
    loop.call_later(5, lambda: print("Прошло 5 секунд"))
 
    # Выполнить callback в конкретное время
    import time
    when = loop.time() + 10  # Через 10 секунд
    loop.call_at(when, lambda: print("Прошло 10 секунд"))
 
    await asyncio.sleep(11)  # Ждём выполнения callbacks
 
# ⚠️ В большинстве случаев вместо этого используйте:
async def better_approach():
    """Более простой и понятный способ."""
 
    await asyncio.sleep(5)
    print("Прошло 5 секунд")  # Проще и понятнее!

async def loop_info():
    """Получение информации о текущем event loop."""
 
    loop = asyncio.get_running_loop()
 
    # Время работы loop (монотонные часы)
    print(f"Loop time: {loop.time()}")
 
    # Проверка, запущен ли loop
    print(f"Is running: {loop.is_running()}")
 
    # Отладка: все активные задачи
    tasks = asyncio.all_tasks(loop)
    print(f"Active tasks: {len(tasks)}")
    for task in tasks:
        print(f"  - {task.get_name()}")

import asyncio
from concurrent.futures import ThreadPoolExecutor
from functools import partial
 
# ❌ Проблема: синхронная библиотека блокирует event loop
def old_sync_code():
    """Синхронный код, который блокирует выполнение."""
 
    import requests
    urls = ["https://example.com"] * 10
 
    results = []
    for url in urls:
        response = requests.get(url)  # Блокирует!
        results.append(response)
    return results
 
# ✅ Решение: используем ThreadPoolExecutor
async def new_async_code():
    """Асинхронная версия с использованием executor."""
 
    import requests
 
    loop = asyncio.get_running_loop()
    urls = ["https://example.com"] * 10
 
    # Создаём pool потоков
    with ThreadPoolExecutor(max_workers=5) as executor:
        tasks = [
            loop.run_in_executor(executor, requests.get, url)
            for url in urls
        ]
 
        # Выполняем параллельно в threads
        responses = await asyncio.gather(*tasks)
 
    return responses
 
# ✅ Ещё лучше: используйте асинхронную библиотеку
async def best_approach():
    """Идеальный вариант с нативной async библиотекой."""
    import aiohttp
 
    urls = ["https://example.com"] * 10
 
    async with aiohttp.ClientSession() as session:
        tasks = [session.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
 
    return responses

import asyncio
 
async def fetch_user(user_id):
    await asyncio.sleep(1)  # Имитация запроса к API
    return {"id": user_id, "name": f"User {user_id}"}
 
async def main():
    # ❌ Плохо: последовательное выполнение
    user1 = await fetch_user(1)  # 1 сек
    user2 = await fetch_user(2)  # ещё 1 сек
    user3 = await fetch_user(3)  # ещё 1 сек
    # Итого: 3 секунды
 
    # ✅ Хорошо: параллельное выполнение
    task1 = asyncio.create_task(fetch_user(1))
    task2 = asyncio.create_task(fetch_user(2))
    task3 = asyncio.create_task(fetch_user(3))
 
    users = await asyncio.gather(task1, task2, task3)
    # Итого: 1 секунда!
    print(users)
 
asyncio.run(main())

import asyncio
 
async def failing_task():
    """Задача, которая падает с ошибкой."""
 
    await asyncio.sleep(0.1)
    raise ValueError("Something went wrong!")
 
# ❌ НЕПРАВИЛЬНО: Исключение будет потеряно
async def bad_example():
    """Исключение НЕ обрабатывается — плохо!"""
 
    task = asyncio.create_task(failing_task())
 
    # Делаем что-то другое
    await asyncio.sleep(1)
 
    # Задача упала, но мы об этом не узнали!
    # В логах будет только:
    # "Task exception was never retrieved"
    print("Программа продолжает работать")
 
asyncio.run(bad_example())

import asyncio
import logging
 
logger = logging.getLogger(__name__)
 
async def fetch_data(url: str) -> dict:
    """Задача, которая может упасть."""
 
    await asyncio.sleep(0.1)
    if url == "bad_url":
        raise ValueError(f"Invalid URL: {url}")
    return {"url": url, "data": "success"}
 
# ✅ ПРАВИЛЬНО: Обрабатываем каждую задачу
async def correct_example_1():
    """Явная обработка исключений для каждой задачи."""
 
    task1 = asyncio.create_task(fetch_data("https://api.com/1"))
    task2 = asyncio.create_task(fetch_data("bad_url"))
    task3 = asyncio.create_task(fetch_data("https://api.com/3"))
 
    # Ждём каждую задачу и обрабатываем ошибки
    for i, task in enumerate([task1, task2, task3], 1):
        try:
            result = await task
            print(f"✅ Task {i}: {result}")
        except ValueError as e:
            logger.error(f"❌ Task {i} failed: {e}")
            # Можем продолжить обработку других задач
        except Exception as e:
            logger.error(f"💥 Task {i} unexpected error: {e}")
 
asyncio.run(correct_example_1())

async def correct_example_2():
    """
    Используем gather с return_exceptions=True.
 
    Преимущество: Не останавливается на первой ошибке.
    """
 
    results = await asyncio.gather(
        fetch_data("https://api.com/1"),
        fetch_data("bad_url"),  # Упадёт с ошибкой
        fetch_data("https://api.com/3"),
        return_exceptions=True  # ← Ключевой параметр!
    )
 
    # Обрабатываем результаты
    for i, result in enumerate(results, 1):
        if isinstance(result, Exception):
            logger.error(f"❌ Task {i} failed: {result}")
        else:
            print(f"✅ Task {i}: {result}")
 
# Вывод:
# ✅ Task 1: {'url': 'https://api.com/1', 'data': 'success'}
# ❌ Task 2 failed: Invalid URL: bad_url
# ✅ Task 3: {'url': 'https://api.com/3', 'data': 'success'}
 
asyncio.run(correct_example_2())

import asyncio
from typing import Set
 
# Хранилище активных задач
background_tasks: Set[asyncio.Task] = set()
 
def handle_task_result(task: asyncio.Task) -> None:
    """
    Callback для обработки результата задачи.
 
    Вызывается автоматически при завершении задачи.
    """
 
    background_tasks.discard(task)  # Убираем из set
 
    try:
        # Получаем результат или исключение
        result = task.result()
        logger.info(f"✅ Task completed: {result}")
 
    except asyncio.CancelledError:
        logger.info("🚫 Task was cancelled")
 
    except Exception as e:
        # Обрабатываем исключение из задачи
        logger.error(f"❌ Task failed: {e}", exc_info=True)
 
# ✅ ПРАВИЛЬНО: Callback обработает исключение
async def correct_example_3():
    """Фоновые задачи с автоматической обработкой ошибок."""
 
    for i in range(5):
        url = "bad_url" if i == 2 else f"https://api.com/{i}"
        task = asyncio.create_task(fetch_data(url))
 
        # Добавляем callback для обработки результата
        task.add_done_callback(handle_task_result)
 
        # Сохраняем ссылку на задачу
        background_tasks.add(task)
 
    # Ждём завершения всех задач
    await asyncio.gather(*background_tasks, return_exceptions=True)
 
asyncio.run(correct_example_3())

import asyncio
import sys
 
if sys.version_info >= (3, 11):
    async def correct_example_4():
        """
        TaskGroup автоматически обрабатывает исключения.
 
        Python 3.11+ рекомендуемый способ!
        """
 
        try:
            async with asyncio.TaskGroup() as tg:
                tg.create_task(fetch_data("https://api.com/1"))
                tg.create_task(fetch_data("bad_url"))
                tg.create_task(fetch_data("https://api.com/3"))
 
        except* ValueError as eg:  # Exception group
            for exc in eg.exceptions:
                logger.error(f"❌ Task failed: {exc}")
 
    # asyncio.run(correct_example_4())

import asyncio
import logging
from typing import List, Callable, Any, TypeVar
 
logger = logging.getLogger(__name__)
 
T = TypeVar('T')
 
async def run_tasks_safely(
    tasks: List[Callable[[], Any]],
    on_error: str = "continue"  # "continue" | "stop" | "collect"
) -> List[Any]:
    """
    Безопасный запуск задач с обработкой ошибок.
 
    Args:
        tasks: Список корутин для выполнения
        on_error: Стратегия при ошибке:
            - "continue": Продолжить выполнение других задач
            - "stop": Остановить при первой ошибке
            - "collect": Собрать все результаты и ошибки
 
    Returns:
        Список результатов (может содержать Exception при on_error="collect")
    """
    task_objects = [asyncio.create_task(task()) for task in tasks]
 
    if on_error == "continue":
        # Продолжаем при ошибках
        results = []
        for task in task_objects:
            try:
                result = await task
                results.append(result)
            except Exception as e:
                logger.error(f"Task failed: {e}", exc_info=True)
                # Не прерываем цикл
        return results
 
    elif on_error == "stop":
        # Останавливаемся при первой ошибке
        results = []
        for task in task_objects:
            result = await task  # Пробросит исключение
            results.append(result)
        return results
 
    else:  # "collect"
        # Собираем все результаты включая ошибки
        return await asyncio.gather(*task_objects, return_exceptions=True)
 
# Использование
async def example():
    tasks = [
        lambda: fetch_data("https://api.com/1"),
        lambda: fetch_data("bad_url"),
        lambda: fetch_data("https://api.com/3")
    ]
 
    # Стратегия 1: Продолжить при ошибках
    results = await run_tasks_safely(tasks, on_error="continue")
    print(f"Получено результатов: {len(results)}")  # 2 (одна упала)
 
    # Стратегия 2: Собрать всё включая ошибки
    all_results = await run_tasks_safely(tasks, on_error="collect")
    for i, r in enumerate(all_results, 1):
        if isinstance(r, Exception):
            print(f"Task {i}: ERROR - {r}")
        else:
            print(f"Task {i}: OK - {r}")
 
asyncio.run(example())

import asyncio
 
async def long_task():
    """Задача с корректной обработкой отмены."""
 
    print("1. Задача запустилась")
 
    try:
        print("2. Начинаю ждать 100 секунд...")
        await asyncio.sleep(100)  # ← Здесь возникнет CancelledError
        print("3. Это НЕ выполнится если отменили")
 
    except asyncio.CancelledError:
        print("4. Получил CancelledError, выполняю cleanup...")
        # Здесь можно закрыть соединения, сохранить состояние и т.д.
        await asyncio.sleep(0.1)  # Cleanup работа
        print("5. Cleanup завершён")
        raise  # ⚠️ ВАЖНО! Пробрасываем дальше
 
    print("6. Это тоже НЕ выполнится при отмене")
 
async def main():
    task = asyncio.create_task(long_task())
 
    # Ждём немного
    await asyncio.sleep(0.5)
 
    # ========================================
    # ШАГ 1: task.cancel() — ЗАПРОС на отмену
    # ========================================
    print(">>> Вызываем task.cancel()")
    task.cancel()
    print(f">>> task.cancelled(): {task.cancelled()}")  # True
    # НО задача ещё работает! Она только помечена для отмены
 
    # ========================================
    # ШАГ 2: await task — ЖДЁМ реальной отмены
    # ========================================
    print(">>> Ждём завершения задачи через await...")
    try:
        await task  # ← Здесь задача реально отменится
        print(">>> Задача завершилась нормально")
    except asyncio.CancelledError:
        print(">>> Поймали CancelledError — задача отменена")
 
    print(f">>> task.done(): {task.done()}")  # True
    print(f">>> task.cancelled(): {task.cancelled()}")  # True
 
asyncio.run(main())

# ❌ НЕПРАВИЛЬНО: Не ждём завершения
async def bad_cancellation():
    task = asyncio.create_task(long_task())
    await asyncio.sleep(0.5)
 
    task.cancel()  # Запросили отмену
    # НО НЕ ДОЖДАЛИСЬ завершения!
 
    # Задача может:
    # 1. Продолжать выполнение cleanup кода
    # 2. Держать открытыми соединения
    # 3. Не освободить ресурсы
    # 4. Вызвать "Task was destroyed but it is pending!"
 
# ✅ ПРАВИЛЬНО: Ждём завершения
async def good_cancellation():
    task = asyncio.create_task(long_task())
    await asyncio.sleep(0.5)
 
    task.cancel()  # Запросили отмену
 
    try:
        await task  # ← Дождались корректного завершения
    except asyncio.CancelledError:
        pass  # Ожидаемо
 
    # Теперь точно знаем что:
    # 1. Cleanup код выполнился
    # 2. Ресурсы освобождены
    # 3. Задача полностью завершена

import asyncio
 
async def detailed_task():
    """Демонстрация поведения при отмене."""
 
    print("Старт")
 
    try:
        # Код ДО первого await выполнится полностью
        print("До await #1")
        x = 10 + 20  # Выполнится
        print(f"x = {x}")
 
        # ← Первый await — ЗДЕСЬ может возникнуть CancelledError
        await asyncio.sleep(1)
        print("После await #1 (может не выполниться)")
 
        await asyncio.sleep(1)  # ← Или здесь
        print("После await #2 (может не выполниться)")
 
    except asyncio.CancelledError:
        print("Отменено! Делаем cleanup...")
        # Здесь можно безопасно использовать await
        await cleanup_resources()
        raise  # Пробрасываем
 
async def cleanup_resources():
    """Cleanup может содержать await."""
    print("Закрываем соединения...")
    await asyncio.sleep(0.1)
    print("Cleanup завершён")
 
async def demo():
    task = asyncio.create_task(detailed_task())
    await asyncio.sleep(0.05)  # Даём задаче чуть-чуть выполниться
 
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Задача корректно отменена")
 
asyncio.run(demo())

# ❌ ПЛОХО: Задача игнорирует отмену
async def stubborn_task():
    try:
        await asyncio.sleep(100)
    except asyncio.CancelledError:
        print("Я не хочу останавливаться!")
        # НЕ пробрасываем raise — ПЛОХО!
        await asyncio.sleep(10)  # Продолжаем работать
        return "I'm still here"  # Задача НЕ отменилась
 
# Попытка отменить
task = asyncio.create_task(stubborn_task())
task.cancel()
result = await task  # Вернёт "I'm still here", а не CancelledError!

task = asyncio.create_task(long_task())
 
task.cancel()  # Первый вызов
task.cancel()  # Второй вызов — ничего не сломается
task.cancel()  # Третий вызов — тоже ок
 
await task  # Дождались один раз

task = asyncio.create_task(some_coroutine())
 
# До отмены
print(task.done())       # False
print(task.cancelled())  # False
 
# После cancel(), но ДО await
task.cancel()
print(task.done())       # False (ещё не завершилась)
print(task.cancelled())  # True (помечена для отмены)
 
# После await
try:
    await task
except asyncio.CancelledError:
    pass
 
print(task.done())       # True (завершена)
print(task.cancelled())  # True (отменена)

import asyncio
import signal
from typing import Set
 
# Глобальное хранилище задач
background_tasks: Set[asyncio.Task] = set()
shutdown_event = asyncio.Event()
 
async def worker(worker_id: int):
    """Рабочая задача."""
    try:
        while not shutdown_event.is_set():
            print(f"Worker {worker_id} working...")
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print(f"Worker {worker_id} cancelling...")
        # Cleanup
        await asyncio.sleep(0.1)
        print(f"Worker {worker_id} stopped")
        raise
 
async def main():
    # Запускаем 5 workers
    for i in range(5):
        task = asyncio.create_task(worker(i))
        background_tasks.add(task)
 
    # Ждём сигнала завершения
    await shutdown_event.wait()
 
    # Graceful shutdown
    print("Shutting down gracefully...")
 
    # Шаг 1: Отменяем все задачи
    for task in background_tasks:
        task.cancel()
 
    # Шаг 2: Ждём их завершения
    results = await asyncio.gather(*background_tasks, return_exceptions=True)
 
    # Шаг 3: Проверяем что все отменились
    for i, result in enumerate(results):
        if isinstance(result, asyncio.CancelledError):
            print(f"Task {i} cancelled successfully")
 
    print("All tasks stopped")
 
# Обработка SIGINT (Ctrl+C)
def handle_sigint():
    print("\nReceived Ctrl+C")
    shutdown_event.set()
 
if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.add_signal_handler(signal.SIGINT, handle_sigint)
 
    try:
        loop.run_until_complete(main())
    finally:
        loop.close()

async def slow_operation():
    await asyncio.sleep(10)
    return "result"
 
async def main():
    try:
        # Ждём максимум 2 секунды
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
    except asyncio.TimeoutError:
        print("Операция заняла слишком много времени!")
 
asyncio.run(main())

async def fetch_data(id):
    await asyncio.sleep(1)
    return f"Data {id}"
 
async def main():
    # Запускаем 5 операций параллельно
    results = await asyncio.gather(
        fetch_data(1),
        fetch_data(2),
        fetch_data(3),
        fetch_data(4),
        fetch_data(5)
    )
    print(results)  # ['Data 1', 'Data 2', ..., 'Data 5']
    # Время: ~1 секунда, не 5!
 
asyncio.run(main())

async def may_fail(id):
    await asyncio.sleep(1)
    if id == 3:
        raise ValueError("Ошибка в задаче 3")
    return f"Result {id}"
 
async def main():
    # return_exceptions=True — не прерывать на ошибке
    results = await asyncio.gather(
        may_fail(1),
        may_fail(2),
        may_fail(3),  # ← Упадёт с ошибкой
        may_fail(4),
        return_exceptions=True  # ← Вернёт Exception вместо raise
    )
 
    for i, result in enumerate(results, 1):
        if isinstance(result, Exception):
            print(f"Задача {i}: ОШИБКА - {result}")
        else:
            print(f"Задача {i}: {result}")
 
asyncio.run(main())

async def task(id, delay):
    await asyncio.sleep(delay)
    return id
 
async def main():
    tasks = [
        asyncio.create_task(task(1, 1)),
        asyncio.create_task(task(2, 2)),
        asyncio.create_task(task(3, 3))
    ]
 
    # Ждём первую завершившуюся задачу
    done, pending = await asyncio.wait(
        tasks,
        return_when=asyncio.FIRST_COMPLETED
    )
 
    print(f"Завершено: {len(done)}")    # 1
    print(f"Ожидают: {len(pending)}")   # 2
 
    # Отменяем оставшиеся
    for task in pending:
        task.cancel()
 
asyncio.run(main())

async def download(url, delay):
    await asyncio.sleep(delay)
    return f"Content from {url}"
 
async def main():
    urls = [
        ("url1.com", 3),
        ("url2.com", 1),
        ("url3.com", 2)
    ]
 
    # Обрабатываем результаты по мере готовности
    for coro in asyncio.as_completed([download(url, delay) for url, delay in urls]):
        result = await coro
        print(result)  # Выведется в порядке: url2, url3, url1
 
asyncio.run(main())

import asyncio
 
class AsyncDatabase:
    async def __aenter__(self):
        print("Подключение к БД...")
        await asyncio.sleep(0.5)  # Имитация подключения
        print("Подключено!")
        return self
 
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Закрытие соединения...")
        await asyncio.sleep(0.2)
        print("Соединение закрыто")
 
    async def query(self, sql):
        await asyncio.sleep(0.1)
        return f"Results for: {sql}"
 
async def main():
    # Автоматическое управление ресурсом
    async with AsyncDatabase() as db:
        result = await db.query("SELECT * FROM users")
        print(result)
    # __aexit__ вызывается автоматически
 
asyncio.run(main())

import aiohttp
 
async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

import asyncio
 
class AsyncRange:
    def __init__(self, start, stop):
        self.current = start
        self.stop = stop
 
    def __aiter__(self):
        return self
 
    async def __anext__(self):
        if self.current >= self.stop:
            raise StopAsyncIteration
 
        await asyncio.sleep(0.5)  # Асинхронная работа
        value = self.current
        self.current += 1
        return value
 
async def main():
    # Асинхронная итерация
    async for number in AsyncRange(0, 5):
        print(number)  # Выводит 0, 1, 2, 3, 4 с задержкой
 
asyncio.run(main())

async def async_range(stop):
    for i in range(stop):
        await asyncio.sleep(0.5)
        yield i
 
async def main():
    async for number in async_range(5):
        print(number)
 
asyncio.run(main())

import asyncio
 
class Counter:
    def __init__(self):
        self.value = 0
        self.lock = asyncio.Lock()
 
    async def increment(self):
        # ✅ Правильно: с блокировкой
        async with self.lock:
            temp = self.value
            await asyncio.sleep(0.01)  # Имитация работы
            self.value = temp + 1
 
    async def broken_increment(self):
        # ❌ Неправильно: гонка данных (race condition)
        temp = self.value
        await asyncio.sleep(0.01)
        self.value = temp + 1
 
async def test():
    counter = Counter()
 
    # Запускаем 100 параллельных инкрементов
    await asyncio.gather(*[counter.increment() for _ in range(100)])
    print(f"С блокировкой: {counter.value}")  # 100 ✓
 
    counter.value = 0
    await asyncio.gather(*[counter.broken_increment() for _ in range(100)])
    print(f"Без блокировки: {counter.value}")  # ~10-50 ✗
 
asyncio.run(test())

import asyncio
 
# Ограничение: максимум 3 одновременных запроса
semaphore = asyncio.Semaphore(3)
 
async def fetch(id):
    async with semaphore:
        print(f"Запрос {id} начат")
        await asyncio.sleep(2)
        print(f"Запрос {id} завершён")
 
async def main():
    # Запускаем 10 запросов, но одновременно выполняются только 3
    await asyncio.gather(*[fetch(i) for i in range(10)])
 
asyncio.run(main())

import asyncio
 
async def waiter(event, name):
    """Корутина, которая ждёт события."""
 
    print(f"{name}: Жду события...")
    await event.wait()  # ← Приостанавливается здесь
    print(f"{name}: Событие произошло!")
 
async def setter(event):
    """Корутина, которая устанавливает событие."""
 
    await asyncio.sleep(2)  # Ждём 2 секунды
    print("🔔 Отправляю событие...")
    event.set()  # ← Разблокирует ВСЕХ ожидающих одновременно
 
async def main():
    event = asyncio.Event()
 
    # Запускаем 3 ожидающих + 1 отправителя
    await asyncio.gather(
        waiter(event, "Задача 1"),
        waiter(event, "Задача 2"),
        waiter(event, "Задача 3"),
        setter(event)
    )
 
asyncio.run(main())

import asyncio
import signal
 
shutdown_event = asyncio.Event()
 
async def worker(worker_id: int):
    """Рабочая задача, которая останавливается по событию."""
 
    while not shutdown_event.is_set():
        print(f"Worker {worker_id} работает...")
        await asyncio.sleep(1)
 
    print(f"Worker {worker_id} завершён")
 
async def main():
    # Запускаем 5 workers
    workers = [asyncio.create_task(worker(i)) for i in range(5)]
 
    # Ждём сигнала завершения (например, Ctrl+C)
    await shutdown_event.wait()
 
    # Все workers автоматически завершатся
    await asyncio.gather(*workers)
 
# В production обрабатываем сигналы:
# signal.signal(signal.SIGINT, lambda s, f: shutdown_event.set())

import asyncio
 
db_ready = asyncio.Event()
 
async def database_connection():
    """Подключение к БД (долгая операция)."""
 
    print("Подключаюсь к БД...")
    await asyncio.sleep(3)  # Имитация подключения
    print("✅ БД готова!")
    db_ready.set()  # Сигнализируем что готовы
 
async def api_handler(request_id: int):
    """API handler, который ждёт готовности БД."""
 
    print(f"Запрос {request_id}: Ожидаю готовность БД...")
 
    # Ждём пока БД инициализируется
    await db_ready.wait()
 
    print(f"Запрос {request_id}: Обрабатываю...")
    await asyncio.sleep(0.5)
    return f"Response {request_id}"
 
async def main():
    # Запускаем всё одновременно
    await asyncio.gather(
        database_connection(),
        api_handler(1),
        api_handler(2),
        api_handler(3)
    )
 
asyncio.run(main())

import asyncio
 
async def reusable_event_example():
    """Пример переиспользования события."""
 
    event = asyncio.Event()
 
    async def waiter():
        for i in range(3):
            print(f"  Раунд {i+1}: Жду...")
            await event.wait()
            print(f"  Раунд {i+1}: Получил!")
            # После wait() НЕ нужно делать clear() вручную
 
    async def controller():
        for i in range(3):
            await asyncio.sleep(1)
            print(f"🔔 Отправляю сигнал {i+1}")
            event.set()
            await asyncio.sleep(0.1)  # Даём обработать
            event.clear()  # ← Сбрасываем для повторного использования
 
    await asyncio.gather(waiter(), controller())
 
asyncio.run(reusable_event_example())

event = asyncio.Event()
event.set()
 
# Все эти wait() вернутся НЕМЕДЛЕННО
await event.wait()  # Не блокируется
await event.wait()  # Не блокируется
await event.wait()  # Не блокируется
 
# Чтобы ждать снова, нужно сбросить:
event.clear()
await event.wait()  # Теперь блокируется

# Кто первый вызвал wait() — не обязательно первый получит управление
# После set() порядок выполнения зависит от event loop scheduler

# ❌ НЕПРАВИЛЬНО: race condition
class BrokenService:
    def __init__(self):
        self.ready = False  # Обычная переменная
 
    async def wait_ready(self):
        while not self.ready:  # Активное ожидание!
            await asyncio.sleep(0.01)  # Жжём CPU
 
# ✅ ПРАВИЛЬНО: используем Event
class CorrectService:
    def __init__(self):
        self.ready = asyncio.Event()
 
    async def wait_ready(self):
        await self.ready.wait()  # Эффективное ожидание

import asyncio
import random
 
async def producer(queue, id):
    """Производитель — добавляет элементы в очередь."""
 
    for i in range(5):
        item = f"P{id}-Item{i}"
        await queue.put(item)
        print(f"📦 Производитель {id} создал: {item}")
        await asyncio.sleep(random.uniform(0.1, 0.5))
 
    print(f"✅ Производитель {id} завершил работу")
 
async def consumer(queue, id):
    """Потребитель — обрабатывает элементы из очереди."""
 
    while True:
        item = await queue.get()
 
        if item is None:  # Сигнал завершения
            queue.task_done()
            break
 
        print(f"  ⚙️  Потребитель {id} обрабатывает: {item}")
        await asyncio.sleep(random.uniform(0.2, 0.4))
 
        queue.task_done()  # ← Обязательно вызываем!
        print(f"  ✓ Потребитель {id} завершил: {item}")
 
async def main():
    # Очередь размером 10 элементов (maxsize ограничивает переполнение)
    queue = asyncio.Queue(maxsize=10)
 
    # Запускаем 2 производителей
    producers = [
        asyncio.create_task(producer(queue, i))
        for i in range(2)
    ]
 
    # Запускаем 3 потребителя
    consumers = [
        asyncio.create_task(consumer(queue, i))
        for i in range(3)
    ]
 
    # Шаг 1: Ждём пока производители закончат создавать задачи
    await asyncio.gather(*producers)
    print("\n🏁 Все производители завершены\n")
 
    # Шаг 2: Ждём пока все задачи из очереди будут обработаны
    await queue.join()
    print("🏁 Все задачи обработаны\n")
 
    # Шаг 3: Отправляем сигналы завершения потребителям
    for _ in consumers:
        await queue.put(None)
 
    # Шаг 4: Ждём завершения потребителей
    await asyncio.gather(*consumers)
    print("🏁 Все потребители завершены")
 
asyncio.run(main())

import asyncio
import aiohttp
 
async def url_producer(queue, urls):
    """Производитель: добавляет URL в очередь."""
 
    for url in urls:
        await queue.put(url)
        print(f"➕ Добавлен URL: {url}")
 
    # Сигналы завершения для всех workers
    for _ in range(3):  # 3 workers
        await queue.put(None)
 
async def scraper_worker(queue, worker_id, session):
    """Потребитель: скачивает страницы."""
 
    while True:
        url = await queue.get()
 
        if url is None:
            queue.task_done()
            break
 
        try:
            print(f"🌐 Worker {worker_id}: загружаю {url}")
            async with session.get(url, timeout=5) as response:
                html = await response.text()
                print(f"✅ Worker {worker_id}: загружено {len(html)} байт")
 
        except Exception as e:
            print(f"❌ Worker {worker_id}: ошибка {url} - {e}")
 
        finally:
            queue.task_done()
 
async def scrape_urls(urls):
    """Главная функция: координирует scraping."""
 
    queue = asyncio.Queue(maxsize=10)  # Буфер на 10 URL
 
    async with aiohttp.ClientSession() as session:
        # Запускаем producer
        producer_task = asyncio.create_task(url_producer(queue, urls))
 
        # Запускаем 3 workers
        workers = [
            asyncio.create_task(scraper_worker(queue, i, session))
            for i in range(3)
        ]
 
        # Ждём завершения
        await producer_task
        await queue.join()
        await asyncio.gather(*workers)
 
# urls = ["https://example.com/page1", ...]
# asyncio.run(scrape_urls(urls))

import asyncio
 
async def batch_processor(queue):
    """Обрабатывает элементы пакетами по 5 штук."""
 
    batch = []
 
    while True:
        item = await queue.get()
 
        if item is None:  # Завершение
            # Обрабатываем остатки
            if batch:
                await process_batch(batch)
            queue.task_done()
            break
 
        batch.append(item)
 
        # Когда накопили 5 элементов — обрабатываем пакет
        if len(batch) == 5:
            await process_batch(batch)
            batch = []
 
        queue.task_done()
 
async def process_batch(batch):
    """Обработка пакета элементов."""
 
    print(f"📦 Обрабатываю пакет из {len(batch)} элементов: {batch}")
    await asyncio.sleep(1)  # Имитация обработки
    print(f"✅ Пакет обработан")

# Обычная очередь (FIFO)
queue.put("Низкий приоритет")  # Обработается первым
queue.put("КРИТИЧНО!")          # Обработается вторым ❌