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

Профилируем Python в продакшене: почему cProfile не подходит, и чем хорош py-spy

Когда на проде начинает течь память или скачет CPU, первая мысль, подключить профайлер. Стандартный cProfile, это детерминированный профайлер. Он хукает каждый вызов функции.

📉 Цена: Оверхед может замедлить приложение в 2-5 раз. На нагруженном проде это означает добить сервис окончательно.

Для Live-систем нужен Sampling Profiler (сэмплирующий профайлер). Золотой стандарт сейчас - py-spy.

Как это работает?
py-spy написан на Rust. Он работает как внешний процесс, который читает память вашего Python-процесса (через системные вызовы, аналогично gdb). Он делает «снимки» стека вызовов с высокой частотой (по дефолту 100 раз в секунду).

✅ Результат: Оверхед стремится к нулю. Код инструментализировать не нужно. Рестарт сервиса не нужен.

Два главных режима работы:

1. Live View (как htop, но для функций)
Посмотреть в реальном времени, в каких функциях процесс проводит больше всего времени.

# Нужно только знать PID процесса
py-spy top --pid 12345

Вы увидите список функций, отсортированный по OwnTime (время внутри функции) и TotalTime (время с учетом дочерних вызовов).

2. Flame Graph (Огненный граф)
Для глубокого анализа лучше записать работу сервиса за период и визуализировать стек.

py-spy record -o profile.svg --pid 12345 --duration 60

Вы получите SVG-файл. Чем шире полоска, тем больше времени занимает функция. Вертикаль - это глубина стека. Сразу видно, кто «съел» процессорное время.

Нюансы для Middle+:

- GIL: py-spy умеет показывать, держит ли функция GIL. Добавьте флаг --gil.
- Docker/K8s: Так как py-spy использует системный вызов ptrace, контейнеру нужны привилегии. В Kubernetes часто нужно добавить securityContext: capabilities: add: ["SYS_PTRACE"] подам, чтобы иметь возможность профилировать их на лету.


#profiling #optimization #pyspy #debugging #python

📲 Мы в MAX

👉@BookPython

Ловушка замыканий: Почему ваши лямбды в цикле сломаны (Late Binding)

Вы пишете код, который генерирует список функций (например, колбэки для кнопок в UI или динамические фильтры). Кажется, что все логично, но на выходе получаете сюрприз.

Проблемный код:

# Хотим создать 3 функции, которые возвращают 0, 1 и 2 соответственно
funcs = []
for i in range(3):
    funcs.append(lambda: i)

# Проверяем
results = [f() for f in funcs]
print(results) 
# Ожидание: [0, 1, 2]
# Реальность: [2, 2, 2]

Почему так происходит?
Это называется Late Binding (позднее связывание).
В Python замыкания (closures) захватывают переменные по ссылке, а не по значению.
Когда вы объявляете lambda: i, Python не сохраняет текущее число 0, 1 или 2. Он сохраняет инструкцию: «когда меня вызовут, пойди в локальную область видимости, найди переменную с именем i и возьми ее значение».

К моменту, когда вы начинаете вызывать функции из списка results, цикл for уже завершился. Переменная i в этой области видимости навсегда осталась равной 2. Все три лямбды смотрят на одну и ту же переменную i.

Как лечить?

Есть два каноничных способа заставить Python запомнить значение «здесь и сейчас».

1. Аргумент по умолчанию (Hack way)
Значения аргументов по умолчанию вычисляются в момент определения функции.

funcs = []
for i in range(3):
    # i=i создает локальную переменную i внутри функции 
    # и присваивает ей текущее значение i из цикла
    funcs.append(lambda i=i: i) 

Это работает быстро, но выглядит немного грязно и может сбить с толку линтеры или коллег.

2. functools.partial (Enterprise way)
Более чистый и явный способ. partial создает новый callable-объект, «замораживая» переданные аргументы.

from functools import partial

funcs = []
for i in range(3):
    # Здесь значение i фиксируется жестко
    funcs.append(partial(lambda x: x, i))

Где это стреляет в реальной жизни?

- Генерация command для кнопок в Tkinter/PyQt.
- Динамическое создание task в asyncio циклах.
- Патчинг тестов в циклах.

Не дайте переменным пережить свое время.

#python #internals #functionalprogramming #gotchas

📲 Мы в 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