Structured Concurrency в Python 3.11+ (TaskGroup vs Gather)

Если вы до сих пор используете asyncio.gather() для запуска конкурентных задач, вы, вероятно, теряете контроль над ошибками.

Главная проблема gather: если одна таска падает, остальные продолжают работать (если не стоит return_exceptions=False, который убивает всё сразу, но "грязно"). Плюс, отлавливать несколько ошибок одновременно через try / except - это боль.

В Python 3.11+ (и через trio / anyio раньше) стандартом стала Structured Concurrency через asyncio.TaskGroup.

Как это выглядит:

import asyncio

async def task_fail(name, sec):
    await asyncio.sleep(sec)
    raise ValueError(f"Error in {name}")

async def task_ok(name, sec):
    await asyncio.sleep(sec)
    print(f"Task {name} done")

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(task_ok("A", 1))
            tg.create_task(task_fail("B", 2)) # Упадет через 2 сек
            tg.create_task(task_fail("C", 1.5)) # Упадет через 1.5 сек
            
    except* ValueError as eg: 
        # Обратите внимание на except* (ExceptionGroup)
        print(f"Caught errors: {eg.exceptions}")

# Запуск
# asyncio.run(main())

Почему TaskGroup это выбор Middle+:

1. Атомарность скоупа: Если одна задача внутри контекстного менеджера падает, TaskGroup автоматически отменяет (cancel) все остальные запущенные задачи в группе. Вы не оставляете "зомби-процессы" в фоне.
2. Exception Groups: Если упало несколько задач одновременно (или во время отмены), Python соберет их в ExceptionGroup. Конструкция except* позволяет элегантно обрабатывать дерево исключений.

Вывод:
Для скриптов gather сойдет. Для надежных сервисов, где важна консистентность состояния и корректный шатдаун корутин - переезжайте на TaskGroup.

#asyncio #python311 #bestpractices #concurrency

📲 Мы в MAX

👉@BookPython

Pydantic V2: Забываем root_validator, используем model_validator правильно

Переход на Pydantic V2, это не только ускорение за счет ядра на Rust, но и переосмысление валидации. Самая частая боль при миграции, проверка зависимостей между несколькими полями.

В V1 мы использовали root_validator и работали со словарем values (прощай, автодополнение IDE). В V2 правильный путь - model_validator в режиме after.

В чем соль?
В режиме mode='after' валидация запускается после того, как поля были распаршены и приведены к типам. Вы работаете с экземпляром класса (self), а не с сырым словарем.

Пример (валидация периода дат):

from pydantic import BaseModel, model_validator
from datetime import datetime

class DateRange(BaseModel):
    start_dt: datetime
    end_dt: datetime

    @model_validator(mode='after')
    def check_dates_order(self):
        # Обращаемся через self — IDE видит поля и их типы!
        if self.end_dt <= self.start_dt:
            raise ValueError("Дата окончания должна быть позже начала")
        return self

# Тест
try:
    DateRange(
        start_dt="2024-01-01T12:00:00", 
        end_dt="2023-01-01T12:00:00"
    )
except ValueError as e:
    print(e)

Нюансы для профи:

1. mode='before': Используйте только если вам нужно модифицировать сырые входные данные (например, JSON) до того, как Pydantic начнет их парсить. Это аналог pre=True из V1.
2. Производительность: Валидаторы на Python, это узкое горлышко. Если у вас HighLoad, старайтесь выразить ограничения через Field (например, ge, le), так как они отрабатывают на Rust-уровне, что значительно быстрее вызова python-функции.

Используйте возможности типизации на 100%.

#pydantic #fastapi #bestpractices #python

📲 Мы в MAX

👉@BookPython

Pytest Patterns: Элегантный Teardown через yield и оптимизация скоупов

Если вы все еще пишете def teardown_method(self): в классах тестов, вы не используете мощь Pytest на 100%.
Фикстуры (fixtures) - это не просто способ передать данные. Это полноценный механизм управления жизненным циклом зависимостей (DI).

1. yield вместо return: Встроенный Teardown
В Pytest фикстура может "замереть", отдать управление тесту, а потом продолжить выполнение. Это реализуется через генератор yield.

Код до yield - это setUp.
Код после yield - это tearDown.

Пример (временная база данных):

import pytest
from sqlalchemy import create_engine

@pytest.fixture
def db_engine():
    # Setup: Поднимаем соединение
    engine = create_engine("sqlite:///:memory:")
    
    # Передаем объект в тест
    yield engine
    
    # Teardown: Этот код выполнится ПОСЛЕ завершения теста
    # (даже если тест упал с ошибкой!)
    engine.dispose()

Это гарантирует, что ресурсы будут освобождены, и вам не нужны try/finally блоки внутри самих тестов.

2. Scopes: Не создавайте мир заново
По умолчанию фикстура имеет scope='function'. Она создается и умирает для каждого теста. Это безопасно, но медленно, если мы говорим о поднятии Docker-контейнера или коннекта к БД.

Используйте scope='session' для тяжелых ресурсов, которые можно переиспользовать.

Паттерн "Изоляция при общем ресурсе":
Частая задача Middle+: иметь одну БД на весь прогон тестов (быстро), но чистые таблицы для каждого теста (изолированно).

Решение: комбинируем скоупы.

# Живет весь прогон тестов (создается 1 раз)
@pytest.fixture(scope="session")
def db_engine():
    engine = create_engine(...)
    yield engine
    engine.dispose()

# Живет 1 тест (создается N раз)
@pytest.fixture(scope="function")
def db_session(db_engine):
    # Берем engine из сессионной фикстуры
    connection = db_engine.connect()
    transaction = connection.begin() # Начали транзакцию
    
    session = Session(bind=connection)
    yield session
    
    session.close()
    # ROLLBACK транзакции после теста вернет базу в исходное состояние!
    transaction.rollback() 
    connection.close()

Итог:

🟢Используйте yield для очистки ресурсов.
🟢Тяжелые объекты (Engine, Client, Container) - в scope='session'.
🟢Легкие объекты с состоянием (Session, User) - в scope='function', наследуясь от тяжелых.

#pytest #testing #qa #bestpractices #python

📲 Мы в MAX

👉@BookPython