Современный Python в 2026: от итераторов до асинхронных контекстных менеджеров

import asyncio
from asyncio import Queue, Semaphore
from typing import Callable, Awaitable, Any
from dataclasses import dataclass
from datetime import datetime
import random

@dataclass
class Task:
    id: str
    data: Any
    priority: int = 0
    created_at: datetime = field(default_factory=datetime.now)
    
    def __lt__(self, other: "Task") -> bool:
        # Для приоритетной очереди
        return self.priority > other.priority

class AsyncWorkerPool:
    """Пул асинхронных воркеров с приоритетной очередью."""
    
    def __init__(
        self,
        worker_count: int,
        handler: Callable[[Task], Awaitable[Any]],
        max_queue_size: int = 1000
    ):
        self.worker_count = worker_count
        self.handler = handler
        self.task_queue: Queue = Queue(maxsize=max_queue_size)
        self.workers: list[asyncio.Task] = []
        self.is_running = False
        self._task_counter = 0
        self._semaphore = Semaphore(worker_count)
        self._metrics = {
            "tasks_processed": 0,
            "tasks_failed": 0,
            "avg_processing_time": 0.0
        }
    
    async def start(self):
        """Запуск воркеров."""
        self.is_running = True
        for i in range(self.worker_count):
            worker = asyncio.create_task(
                self._worker_loop(f"worker-{i}"),
                name=f"worker-{i}"
            )
            self.workers.append(worker)
        print(f"Started {self.worker_count} workers")
    
    async def stop(self):
        """Остановка воркеров."""
        self.is_running = False
        # Ожидаем завершения всех задач
        await self.task_queue.join()
        # Отменяем воркеров
        for worker in self.workers:
            worker.cancel()
        await asyncio.gather(*self.workers, return_exceptions=True)
        print("All workers stopped")
    
    async def submit_task(self, data: Any, priority: int = 0) -> str:
        """Добавление задачи в очередь."""
        task_id = f"task-{self._task_counter}"
        self._task_counter += 1
        
        task = Task(id=task_id, data=data, priority=priority)
        await self.task_queue.put(task)
        return task_id
    
    async def _worker_loop(self, worker_name: str):
        """Цикл обработки задач воркером."""
        while self.is_running:
            try:
                task = await self.task_queue.get()
                
                start_time = datetime.now()
                try:
                    async with self._semaphore:
                        result = await self.handler(task)
                    processing_time = (datetime.now() - start_time).total_seconds()
                    
                    # Обновляем метрики
                    self._metrics["tasks_processed"] += 1
                    total_time = self._metrics["avg_processing_time"] * \
                                (self._metrics["tasks_processed"] - 1)
                    self._metrics["avg_processing_time"] = \
                        (total_time + processing_time) / self._metrics["tasks_processed"]
                    
                    print(f"{worker_name} processed {task.id} in {processing_time:.3f}s")
                    
                except Exception as e:
                    self._metrics["tasks_failed"] += 1
                    print(f"{worker_name} failed to process {task.id}: {e}")
                
                finally:
                    self.task_queue.task_done()
            
            except asyncio.CancelledError:
                break
    
    def get_metrics(self) -> dict:
        """Получение текущих метрик."""
        return self._metrics.copy()

async def example_handler(task: Task) -> str:
    """Пример обработчика задач."""
    await asyncio.sleep(random.uniform(0.1, 0.5))  # Имитация работы
    return f"Processed {task.id} with data: {task.data}"

async def main_worker_pool():
    """Демонстрация работы пула воркеров."""
    pool = AsyncWorkerPool(
        worker_count=5,
        handler=example_handler,
        max_queue_size=100
    )
    
    await pool.start()
    
    # Добавляем задачи
    tasks = []
    for i in range(20):
        task_id = await pool.submit_task(
            data=f"payload-{i}",
            priority=random.randint(1, 10)
        )
        tasks.append(task_id)
    
    # Даем время на обработку
    await asyncio.sleep(2)
    
    # Смотрим метрики
    print(f"Metrics: {pool.get_metrics()}")
    
    await pool.stop()

4.2. Асинхронный Event Bus для микросервисов

from typing import Type, TypeVar, Generic, Callable, Set
from enum import Enum
import asyncio
from dataclasses import dataclass
from abc import ABC, abstractmethod

T = TypeVar('T')

class EventType(Enum):
    USER_CREATED = "user.created"
    ORDER_PLACED = "order.placed"
    PAYMENT_PROCESSED = "payment.processed"
    SYSTEM_ALERT = "system.alert"

@dataclass
class Event(Generic[T]):
    """Базовый класс события."""
    event_type: EventType
    data: T
    timestamp: float = None
    correlation_id: str = None
    
    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = asyncio.get_event_loop().time()

class EventHandler(ABC, Generic[T]):
    """Абстрактный обработчик событий."""
    
    @abstractmethod
    async def handle(self, event: Event[T]) -> None:
        pass

class EventBus:
    """Асинхронная шина событий."""
    
    def __init__(self):
        self._handlers: dict[EventType, Set[EventHandler]] = {}
        self._queue: asyncio.Queue = Queue(maxsize=10000)
        self._dispatcher_task: asyncio.Task = None
    
    def subscribe(self, event_type: EventType, handler: EventHandler):
        """Подписка на события."""
        if event_type not in self._handlers:
            self._handlers[event_type] = set()
        self._handlers[event_type].add(handler)
    
    def unsubscribe(self, event_type: EventType, handler: EventHandler):
        """Отписка от событий."""
        if event_type in self._handlers:
            self._handlers[event_type].discard(handler)
    
    async def publish(self, event: Event):
        """Публикация события."""
        await self._queue.put(event)
    
    async def start(self):
        """Запуск диспетчера событий."""
        self._dispatcher_task = asyncio.create_task(self._dispatch_events())
    
    async def stop(self):
        """Остановка диспетчера."""
        if self._dispatcher_task:
            self._dispatcher_task.cancel()
            try:
                await self._dispatcher_task
            except asyncio.CancelledError:
                pass
    
    async def _dispatch_events(self):
        """Диспетчер событий."""
        while True:
            try:
                event = await self._queue.get()
                
                if event.event_type in self._handlers:
                    handlers = self._handlers[event.event_type].copy()
                    
                    # Запускаем обработчики конкурентно
                    tasks = []
                    for handler in handlers:
                        task = asyncio.create_task(
                            handler.handle(event),
                            name=f"handler-{handler.__class__.__name__}"
                        )
                        tasks.append(task)
                    
                    # Ожидаем завершения всех обработчиков
                    await asyncio.gather(*tasks, return_exceptions=True)
                
                self._queue.task_done()
                
            except asyncio.CancelledError:
                break
            except Exception as e:
                print(f"Error in event dispatcher: {e}")

# Пример использования
@dataclass
class UserData:
    user_id: int
    username: str
    email: str

class UserCreatedHandler(EventHandler[UserData]):
    """Обработчик создания пользователя."""
    
    async def handle(self, event: Event[UserData]) -> None:
        print(f"Handling user created: {event.data.username}")
        # Отправка приветственного email
        await asyncio.sleep(0.1)
        # Создание профиля
        await asyncio.sleep(0.1)
        print(f"User {event.data.user_id} setup completed")

class AnalyticsHandler(EventHandler[UserData]):
    """Обработчик для аналитики."""
    
    async def handle(self, event: Event[UserData]) -> None:
        print(f"Analytics: new user {event.data.user_id}")
        await asyncio.sleep(0.05)

async def main_event_bus():
    """Демонстрация работы шины событий."""
    bus = EventBus()
    
    # Регистрируем обработчики
    user_handler = UserCreatedHandler()
    analytics_handler = AnalyticsHandler()
    
    bus.subscribe(EventType.USER_CREATED, user_handler)
    bus.subscribe(EventType.USER_CREATED, analytics_handler)
    
    # Запускаем шину
    await bus.start()
    
    # Публикуем события
    for i in range(5):
        user_event = Event[UserData](
            event_type=EventType.USER_CREATED,
            data=UserData(
                user_id=i + 1,
                username=f"user_{i}",
                email=f"user_{i}@example.com"
            ),
            correlation_id=f"corr-{i}"
        )
        await bus.publish(user_event)
        await asyncio.sleep(0.1)
    
    # Даем время на обработку
    await asyncio.sleep(1)
    
    # Останавливаем шину
    await bus.stop()

Заключение

Современный Python в 2026 году — это мощный инструмент для построения сложных, высоконагруженных систем. Ключевые тренды, которые мы рассмотрели:

✅ Итераторы и генераторы перешли от простых циклов к сложным stateful-паттернам обработки потоков данных
✅ Контекстные менеджеры эволюционировали в полноценные системы управления ресурсами с поддержкой транзакций, circuit breakers и health checks
✅ Декораторы стали инструментом для внедрения cross-cutting concerns: кеширования, валидации, retry-логики
✅ Асинхронные паттерны (worker pools, event buses) позволяют строить системы, обрабатывающие миллионы событий в секунду
✅ Метапрограммирование используется для создания DSL, ORM и автоматической кодогенерации

Главная сила Python сегодня — в балансе между производительностью и выразительностью. Вы можете начать с простого скрипта и постепенно, используя те же языковые конструкции, развить его до распределенной системы уровня enterprise.