About Python [ru]

🤣 Нейроновый мир все таки победил?) ✖️ xCode Journal

Java в 2026: куда движется язык - узнай на вебинаре! Java в 2026: разберём тренды, ниши и реальные вакансии - поймёте, куда расти как разработчику. Запишитесь сейчас! Узнать больше #реклама 16+ otus.ru О рекламодателе

⁣contextvars и asyncio.Lock: сквозной трейсинг с гарантированной очисткой Типичная проблема в асинхронных middleware-цепях (FastAPI, aiohttp, Sanic) — race condition при записи в контекст: несколько корутин перезатирают request_id друг друга, а утечка контекста после ошибки ломает последующие запросы. Решение — contextvars.ContextVar с asyncio.Lock для атомарности и finally для гарантированной очистки. Датакласс TraceContext с блокировкой Храним request_id, user_id и встроенный asyncio.Lock. Lock сериализует доступ к общему состоянию внутри одной middleware-цепи, исключая гонки.

from contextvars import ContextVar
from dataclasses import dataclass, field
import asyncio

@dataclass
class TraceContext:
    request_id: str = ''
    user_id: int | None = None
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock, compare=False)

current_trace = ContextVar('current_trace', default=TraceContext())

Middleware с тайм-аутом и очисткой Каждый middleware делает set() один раз, Lock гарантирует, что параллельные корутины не испортят контекст. finally сбрасывает токен — утечка исключена. Тайм-аут в 30 секунд обрезает зависшие цепочки.

async def tracing_middleware(request, call_next):
    token = current_trace.set(TraceContext(request_id=str(uuid4())))
    try:
        async with asyncio.timeout(30):
            async with current_trace.get()._lock:
                return await call_next(request)
    except asyncio.TimeoutError:
        raise
    finally:
        current_trace.reset(token)

Чтение без прокидывания AuthMiddleware и LoggingMiddleware читают current_trace.get() — никаких параметров через request.state или сигнатуру. Lock не даёт двум параллельным задачам испортить один trace: если одна корутина ждёт I/O, другая не перезатрёт request_id. Вывод: Связка contextvars + asyncio.Lock в middleware даёт потокобезопасный контекст с гарантированной очисткой и тайм-аутом, устраняя race conditions и утечки в асинхронных production-сервисах.

⁣Микрооптимизация сериализации в JSON/msgpack: от скрытых словарей до кастомных кодеков Под high-throughput RPC стандартная сериализация через __dict__ или рефлексию превращается в узкое место. Многие разработчики забывают, что каждый вызов json.dumps или msgpack.packb с объектом по умолчанию парсит всю структуру через __dict__, добавляя лишние накладные расходы. __slots__: убиваем __dict__ на старте У обычного класса атрибуты хранятся в __dict__- словаре с хеш-таблицей, замедляющей доступ. __slots__ убирает этот словарь, поля читаются напрямую. Для сериализации это даёт двойной выигрыш: меньше аллокаций при packb/unpackb и более быстрый доступ к значениям. * Класс с __slots__ занимает меньше памяти, что критично при сотнях тысяч объектов. * Внутри msgpack можно напрямую читать поля без итерации по __dict__. * Торгуем гибкостью динамических атрибутов на производительность.

import msgpack

class Point:
    __slots__ = ('x', 'y')
    def __init__(self, x, y):
        self.x = x
        self.y = y

p = Point(1, 2)
data = msgpack.packb([p.x, p.y], use_bin_type=True)  # минимум накладных расходов

__reduce__: явное описание структуры для msgpack Обычно __reduce__ используют для pickle, но его можно адаптировать под msgpack. Метод возвращает кортеж с идентификатором класса и данными, убирая всю рефлексию при парсинге.

class MyData:
    __slots__ = ('a', 'b')
    def __reduce__(self):
        return (self.__class__, (), {'a': self.a, 'b': self.b})

data = msgpack.packb(obj, default=lambda o: o.__reduce__(), use_bin_type=True)

Этот подход позволяет избежать стандартного вызова __dict__ и даёт прямой доступ к полям. Однако будьте осторожны: если структура данных меняется, __reduce__ может вернуть устаревший кортеж, и десериализация сломается. Кастомные кодеки: контроль над упаковкой В msgpack есть ExtType, в JSON - __json__ метод или кастомный encoder. Кастомный кодек позволяет упаковать объект в минимальный набор данных, например, числа в бинарные строки или сложные структуры в один вызов packb.

import msgpack

class FastEncoder:
    ext_type = 42
    def encode(self, obj):
        if isinstance(obj, MyFastClass):
            return msgpack.packb([obj.x, obj.y], use_bin_type=True)
        raise TypeError

msgpack.packb(obj, default=FastEncoder().encode, strict_types=True)

На практике, под high-throughput RPC, комбинация __slots__ + __reduce__ + кастомный кодек даёт выигрыш в пропускной способности до 20-30% на мелких объектах. Но если объекты содержат много вложенных полей или тяжёлые структуры (например, строки длиннее 100 символов), кастомные кодеки могут стать медленнее из-за дополнительных вызовов packb. Торгуйте производительность и читаемость: для простых DTO - этот подход, для сложных графов - стандартная сериализация с оптимизацией в виде orjson. Вывод: В high-throughput RPC микрооптимизации сериализации с __slots__, __reduce__ и кастомными кодеками дают реальный прирост производительности, но требуют строгого контроля над форматом данных и готовности к увеличению сложности поддержки.

Директ возвращает 10% бюджета за рекламу в мессенджерах Запускайте кампании в Telegram и МАКС с 1 июня по 31 августа — получайте кешбэк 10%. Его можно потратить на новые кампании. 💰 Как получить кешбэк: — Заполните форму: укажите логин в Директе и что продвигаете (сайт или канал); — Запустите рекламу в МАКС, Telegram или сразу на двух площадках с оплатой за клики; — В сентябре начислим кешбэк. Получить предложение #реклама yandex.ru О рекламодателе

DTO, schema, model, entity: почему в коде всё называется User Один класс User может использоваться для приёма запроса, выдачи в API, сохранения в БД и передачи между сервисами. Со временем он смешивает разные границы, и код перестаёт защищать от ошибок. В статье на Python показаны различия между DTO, schema, model и entity. Отдельные классы разграничивают ответственность: DTO для передачи данных между слоями, schema для валидации входящих/исходящих данных, model для работы с БД, entity для бизнес-логики. Когда они не разделены, изменение в одной части кода ломает другие. Примеры демонстрируют, как выделение отдельных классов упрощает поддержку и защищает от неожиданных изменений. Читать далее

Утраиваем бюджет на продвижение в Директе Запустите первое продвижение в Яндекс Директе с утроенным бюджетом и ИИ-помощником ✨ Используйте один из промокодов : При пополнении от 10 000 ₽ +20 000 ₽ Промокод START20 При пополнении от 15 000 ₽ +30 000 ₽ Промокод START30 Зарегистрироваться #реклама direct.yandex.ru О рекламодателе

⁣Асинхронные краны сообщений через asyncio.Queue с приоритетами и тайм-аутами: диспетчер задач без Celery Когда Celery — это перебор, а Redis-очередь лень поднимать, многие кидаются на asyncio.Queue. Но голая FIFO не отдаст приоритет срочным задачам, и зависший воркер подвесит всю систему. Решение — собственный диспетчер с приоритетами и тайм-аутами. Датакласс с сортировкой Используем @dataclass(order=True), чтобы задачи сортировались по приоритету автоматически. Поле timeout задаёт лимит на выполнение, а payload — полезная нагрузка. Это даёт чистый интерфейс без внешних зависимостей. Воркер с тайм-аутом Обёртка asyncio.wait_for в цикле воркера режет задачу по тайм-ауту. Если не уложилась — кидаем TimeoutError, но очередь не ломается, и воркер спокойно переходит к следующей. Это ключевой приём для production, где один зависший IO-запрос не должен блокировать остальные. Production-пример В реальном проекте (обработка пайплайна алертов) приоритеты определяют важность: priority=1 для критических, priority=5 для отчётов. Тайм-аут на 2 секунды не даёт медленному внешнему API тянуть всю очередь. Код минимален и работает на Python 3.7+.

import asyncio
from dataclasses import dataclass, field

@dataclass(order=True)
class Task:
    priority: int
    timeout: int = field(default=10, compare=False)
    payload: str = field(default=None, compare=False)

async def worker(queue: asyncio.PriorityQueue, name: str):
    while True:
        task: Task = await queue.get()
        try:
            await asyncio.wait_for(process(task.payload), timeout=task.timeout)
        except asyncio.TimeoutError:
            print(f"[{name}] Timed out priority {task.priority}")
        finally:
            queue.task_done()

Типичная ошибка Начинающие забывают про queue.task_done() — это ведёт к зависанию queue.join(). Или не оборачивают задачу в asyncio.wait_for, тогда одна долгая операция блокирует всех воркеров. Трейд-оффы Вся очередь живёт в памяти — упадёт процесс, задачи потеряны. Нет распределённости: воркеры работают в одном процессе. На высоких нагрузках (10k+ задач/сек) вставка O(log n) через heapq может просадить производительность. Для повышения надёжности добавляйте asyncio.Semaphore для лимита параллельных задач и retry с повышением приоритета. Вывод: Для простых однопроцессных сценариев с приоритетами и тайм-аутами asyncio.PriorityQueue — работающий лёгкий инструмент, но без персистентности и распределённости он не конкурент Celery на кластерных нагрузках.

Телеграм канал AI для бизнеса AI - не будущее. Это настоящее вашего бизнеса. Телеграм-канал "AI для бизнеса" знает все о внедрении и использовании искусственного интеллекта в бизнесе в России и мира. Только со своими подписчиками канал делится: - как внедрить искусственный интеллект в реальные бизнес-процессы, - разборами кейсов: как компании сократили затраты на 30-50% с помощью AI, - лайфхаками по автоматизации рутинных задач, - новостями мира AI и разборами трендов. Сами давно читаем и вам советуем подписаться. Подписаться #реклама 16+ О рекламодателе

Как я устал писать парсер под каждый прайс и сделал из этого библиотеку На проекте десятки прайсингов на топливо: один вендор шлёт CSV, другой Excel, третий JSON на вебхук. Данные одни, но колонка цены везде называется по-своему, даты в трёх форматах, единицы то литры, то галлоны, а половина нужных полей отсутствует. Под каждый источник жил отдельный парсер на сотню строк if-else. Сначала их было три, потом восемь, потом количество перестали считать. Парсеры ломались молча: вендор тихо переименовывал колонку. В третий раз за месяц копируя один и тот же парсер, я понял, что так нельзя, и вынес логику маппинга из кода в данные. Из этого выросла библиотека fidelis: данные описываются один раз как Pydantic-модель, а соответствие под каждый кривой источник один раз пишет LLM — в виде читаемой YAML-спеки, которую ревьюят и коммитят. Дальше LLM не нужен: чистый детерминированный Python, валидация каждой строки и отлов изменений схемы ещё в CI. Источник

Получи грант до 3,48 млн на обучение дизайну Поступай на дизайн в Центральный университет с грантом. Для учеников 10–11-х классов и СПО. Освой графический, UI/UX и продуктовый дизайн. Создавай визуальные концепты будущего. На программе студенты получают фундаментальную базу, развивают прикладные навыки, приобретают опыт работы над реальными проектами, собирают портфолио и строят связи внутри дизайн-сообщества Подать заявку #реклама 16+ cu.ru О рекламодателе

⁣Профилирование утечек памяти в asyncio: трассировка task-стека и gc.get_objects Утечки памяти в asyncio-приложениях — та ещё головная боль. Стандартные pympler или objgraph часто показывают не то, потому что объекты висят в стеках корутин или в циклических ссылках внутри Task-ов. Есть два рабочих подхода, которые я сам использую. Трассировка стека через Task.get_stack() Когда таск зависает в бесконечном ожидании (например, из-за Future, который никогда не завершится), его кадры стека держат ссылки на большие объекты. Берёшь asyncio.all_tasks(), для каждого вызываешь get_stack(), ищешь кадры с подозрительными локальными переменными. Прямо видишь, что висит и сколько весит:

async def leaky_task():
    data = [0] * 10_000_000
    await asyncio.Future()

tasks = asyncio.all_tasks()
for t in tasks:
    stack = t.get_stack()
    if stack and 'data' in stack[0].f_locals:
        obj = stack[0].f_locals['data']
        print(f'Task {t} держит {type(obj)} размером {sys.getsizeof(obj)} байт')

Поиск забытых ссылок через gc.get_objects Снимаешь снапшот до операции, потом после, ищешь разницу. Только не забудь вызвать gc.collect() перед каждым снимком, иначе поймаешь кучу короткоживущих объектов.

def find_growth(snapshot_before):
    new_objects = []
    for obj in gc.get_objects():
        if id(obj) not in snapshot_before and hasattr(obj, '__class__'):
            new_objects.append(obj.__class__.__name__)
    return Counter(new_objects).most_common(5)

Почему это работает? В asyncio event loop держит ссылки на все незавершённые таски. Если корутина содержит локальные переменные или замыкания — они не утилизируются, пока таск не завершится. gc.get_objects находит объекты, которые держатся циклическими ссылками в колбэках или корутинах. Практические советы: - Включи gc.set_debug(gc.DEBUG_SAVEALL) — он сохраняет недостижимые объекты, можно посмотреть, кто не собирается. - Для единичных подозрительных корутин sys.getrefcount до и после — быстро, но грубо. - Для продакшена есть tracemalloc, который не требует остановки приложения. Типичная ошибка: Не вызывать gc.collect() перед снимком — тогда разница будет забита короткоживущими объектами из недавних операций, что маскирует реальную утечку. Вывод: Трассировка стека тасков ловит утечки внутри активных корутин, а gc.get_objects — глобальные забытые объекты и циклические ссылки, и без них вы будете сидеть с логами и гадать, где память утекает.

Онлайн-магистратура для IT: ИТМО, МИФИ + Яндекс Программы онлайн-магистратуры ИТМО и МИФИ в партнёрстве с Яндексом. Актуальные знания, практическое обучение и гибкий график. Учитесь, совмещая с работой. Доступна господдержка оплаты, отсрочка от армии Перейти на сайт #реклама 16+ practicum.yandex.ru О рекламодателе

⁣PEP 738: __main__.py как исполняемый zip-архив — деплой без интерпретатора в scratch-образы Долгое время деплой Python-сервисов в Docker означал обязательную установку интерпретатора, даже slim-образы весят 50-100+ МБ. Для микросервисов это расточительно и увеличивает время развертывания. PEP 738, реализованный в Python 3.12+, меняет подход, позволяя использовать zipapp для создания исполняемых архивов. Как это работает PEP 738 разрешает исполнять zip-архив, содержащий __main__.pypython app.zip. В основе лежит класс zipimport, который теперь поддерживает загрузку кода без распаковки. Для деплоя в scratch-образ нужно скомпилировать Python в статический бинарник. * Упакуйте приложение с зависимостями:

# project/
#   __main__.py
#   app/
#   vendor/
python -m zipapp project -o app.zip

* Используйте python-build-standalone или nuitka для получения статического бинарника без внешних зависимостей. * Итоговый Dockerfile:

FROM scratch
COPY static-python /python
COPY app.zip /
ENTRYPOINT ["/python", "/app.zip"]

Production-пример: HTTP-микросервис Рассмотрим микросервис на FastAPI:

# __main__.py
import uvicorn
from app.main import app
uvicorn.run(app, host="0.0.0.0", port=8000)

Сборка:

python -m zipapp project -o app.zip
# Статический бинарник Python ~5-8 МБ
FROM python:3.12-slim AS builder
COPY app.zip .
FROM scratch
COPY --from=builder /python /python
COPY --from=builder /app.zip /
ENTRYPOINT ["/python", "/app.zip"]

Итоговый образ весит 10-20 МБ вместо 100+ МБ. Это напрямую ускоряет деплой в Kubernetes и уменьшает трафик при пуле образов. Типичная ошибка и ограничения * Ошибка: Попытка включить нативные C-расширения вроде psutil или cryptography в архив. Они не работают без /usr/lib — нужна статическая сборка этих библиотек. * Предупреждение: Не все библиотеки совместимы. Например, pandas или numpy могут требовать системных .so-файлов. Выход — использовать manylinux-совместимые wheels или статические билды. Практический совет и trade-offs Для сервисов на asyncio с простыми зависимостями (HTTP-клиенты, JSON, SQLAlchemy без C-ускорений) этот подход идеален. Но для CLI-утилит или Lambda-функций он уже стандарт. Главный trade-off: уменьшение размера образа на 80% за счет невозможности использовать системные утилиты и сложность отладки (нет bash, curl). Для observability используйте только exec-форму ENTRYPOINT. Вывод: PEP 738 и zipapp позволяют сократить размер Docker-образов для Python-микросервисов до 10-20 МБ, жертвуя гибкостью системного окружения ради производительности деплоя и безопасности scratch-образов.

⁣Стратегия фрагментации и дефрагментации памяти в Python: управление аллокацией через pymalloc и кастомные арены под long-running сервисы Когда сервис живёт неделями, память ведёт себя как кошка — вроде есть, но непонятно где. pymalloc дробит память на пулы (4KB) и арены (256KB), чтобы реже дёргать malloc/free, но не дефрагментирует пулы сам. Освободил блоки — они висят, пока в пуле жив хотя бы один объект. После пика нагрузки получаешь 256KB арену с 20% занятости, а RSS растёт без причины. Диагностика: что посмотреть Смотри sys.getpymallocstats() (собранное с флагом) или используй tracemalloc и memray. Но это диагностика, не лечение. Для прода — sys.getallocatedblocks и мониторинг RSS, чтобы понять, когда память утекает не в объекты, а во внутреннюю фрагментацию. Разделяй данные: долгоживущие и временные Если у тебя кеш, который висит всё время, не мешай его с временными объектами в одних аренах. Выделяй под кеш память через mmap вручную — тогда pymalloc вообще не трогает эти регионы. Пример для production:

import mmap
buf = mmap.mmap(-1, 1048576, prot=6)  # 1 MB, RW

Торгов: проигрыш в аллокации на уровне ОС, но выигрыш в предсказуемости и отсутствии фрагментации. Ручной сброс: почему это не панацея Удалил ссылки на временные объекты — GC сработал, pymalloc освободил блоки внутри пулов. Но пул вернётся в систему только когда он полностью пуст. Арена — когда пусты все пулы. Гарантий возврата памяти ОС нет, пока процесс жив. Типичная ошибка: ждать, что после gc.collect() RSS упадёт — не упадёт, если хоть один объект держит арену. Экзотика для упоротых PYTHONMALLOC=malloc отключает pymalloc, но даёт overhead на каждый мелкий объект. На Linux mallopt(M_MMAP_THRESHOLD) управляет системными вызовами. Для самых требовательных — кастомные аллокаторы через ctypes или C extension, но это edge-case для high-frequency trading. Практический совет для long-running сервиса Не надеяться, что Python сам вернёт память. Мониторить RSS и sys.getallocatedblocks. Проектировать логику так, чтобы можно было пересоздать пулы целиком — перезагрузить часть данных, сбросить кеш и дать аренам освободиться. pymalloc хорош, но он про скорость, не про экономию памяти подолгу. Вывод: pymalloc оптимизирован под скорость аллокации мелких объектов, а не под долговременное удержание памяти — в long-running сервисах осознанное разделение арен и мониторинг фрагментации критичнее, чем надежда на автоматическую дефрагментацию.

⁣Immutable Config, Hot-Reload и Версионирование: Стратегия управления конфигурацией в Python-сервисах Конфиги часто воспринимаются как статические файлы, которые загружают и забывают. Однако в production именно мутация конфига или невалидные данные приводят к трудноотловимым багам, гонкам данных и внезапным падениям после деплоя. Разберем три практики, которые делают конфигурацию надежной в Python 3.11+. Immutable Config после загрузки После инициализации конфиг не должен изменяться. Избегайте глобальных словарей, которые кто-то может дописать по ходу работы. Используйте Pydantic v2 с frozen=True или dataclass с slots. Это защищает от случайной мутации и гарантирует, что все зависимости от конфига будут зафиксированы на старте. В асинхронных сервисах это критично — одновременное чтение и запись из нескольких корутин легко ломает логику. Hot-reload через inotify без боли Для сервисов с uptime 24/7 перезапуск ради смены таймаута или уровня логирования излишен. Используйте watchdog, который через inotify отслеживает изменения файла конфига. Вешайте FileSystemEventHandler на конкретный файл, а не на директорию, чтобы избежать ложных срабатываний. Типичная ошибка: обработчик срабатывает на незаконченную запись, когда файл очищается перед перезаписью. Решение — загружать конфиг атомарно (через write-and-rename) или проверять, что в конфиге есть все обязательные поля, через try-except. Версионирование схем для обратной совместимости Новая версия сервиса не должна падать из-за старого конфига или наоборот. Используйте наследование от базовой модели ConfigV1, где поле version — обязательное. При загрузке сверяйте версию и подставляйте нужную модель-наследник с добавленными полями. В Pydantic v2 параметр extra="forbid" отсекает мусорные ключи, а строгая валидация типов ловит несоответствия еще до старта приложения. Пример: class ConfigV2(ConfigV1): new_field: str. Практический совет Храните файлы конфигов отдельно от кода (например, в Consul или AWS Parameter Store), а в CI добавьте шаг валидации через pydantic + pyyaml. Это предотвратит попадание невалидных конфигов в прод даже до деплоя. Предупреждение Не смешивайте конфиги с переменными окружения без структуры: pydantic-settings для env хорош, но забывают, что переменные окружения — тоже часть конфигурации, и их необходимо версионировать вместе со схемой. Вывод: Неизменяемость конфига после загрузки, атомарный hot-reload и версионирование схем через Pydantic v2 — три кита, которые делают конфигурацию предсказуемой и защищают production от сбоев, связанных с некорректными данными.

Django-кнопка «Наверх»: готовое решение для проектов Кнопка «Наверх» — вещь настолько простая, что обычно её делают прямо в проекте: ссылка, немного CSS, пара строк JavaScript — готово. Для одной страницы это нормально. Но потом проект растёт. Появляется мобильная версия, cookie-баннер, чат в углу, требования к контрасту, строгий CSP. Где-то нужно добавить кнопку ещё и в Django Admin. И тот самый маленький кусок кода начинает жить своей жизнью. Такой модуль нужен давно: кнопка «Наверх» встречается на множестве сайтов, но в Django-проектах её по-прежнему часто собирают вручную — каждый раз немного по-своему. Мне хотелось получить готовое решение: подключил, настроил через админку и больше не возвращаешься к этому коду при каждом новом проекте. Так появился django-scroll-to-top. Читать далее

⁣PEP 723: Inline Script Metadata — упаковка однофайловых скриптов без Poetry/uv У вас есть скрипт на сервере: дёргает API, пишет в базу. Без venv, без poetry — все ставят pip install requests и молятся, что версия та. PEP 723 решает эту проблему, вписывая зависимости прямо в скрипт через блок-комментарий с метаданными. Как это работает Структура проста: блок # /// script до # /// содержит TOML-синтаксис, знакомый по pyproject.toml. Утилита pip-run парсит его, создаёт изолированное окружение и запускает скрипт. Начиная с Python 3.11 поддержка встроена, но pip-run удобнее в продакшене.

# /// script
# requires-python = ">=3.11"
# dependencies = [
#   "requests>=2.31.0",
#   "click>=8.1.0"
# ]
# ///

import requests, click

@click.command()
@click.option('--url', required=True)
def main(url):
    response = requests.get(url)
    print(f"Status: {response.status_code}")

if __name__ == "__main__":
    main()

Production-пример: ETL-скрипт на агенте мониторинга Используйте это для ETL-задач или метрик: скопировали файл на свежую машину, запустили pip-run script.py — и всё работает. Никаких конфликтов версий с другими проектами, никакого ручного создания venv. Это идеально для CI/CD и агентов, где каждый скрипт живёт сам по себе. Типичная ошибка Не пытайтесь использовать это для большой codebase — это не замена классическому менеджеру пакетов. pip-run не поддерживает lock-файлы и разрешение зависимостей на уровне проекта. Для одного скрипта это нормально, но для целого сервиса с десятками файлов — оставьте poetry или uv. Практический совет Проверяйте совместимость Python-версии: если окружение агента — Python 3.9, укажите requires-python = ">=3.9". Иначе pip-run молча выберет последнюю версию, что сломает совместимость с системными библиотеками. Вывод: PEP 723 избавляет от магии в зависимостях однофайлового продакшена, но не заменяет полноценный менеджер пакетов для сложных проектов.

⁣Генерация плоских protobuf-контейнеров через __init_subclass__ и метаклассы: Zero‑allocation DTO без Pydantic под high‑flow RPC Когда RPC-сервис пережевывает миллионы коротких запросов в секунду, каждая лишняя аллокация — боль. Pydantic под капотом создает словари, кэши, трейсы — для обычного API норм, но для high-flow это убивает latency. Типичная ошибка: использовать универсальный DTO, не задумываясь о цене каждого байта. Проблема лишних оберток Protobuf-сообщения уже имеют слоты и фиксированный размер. Но после десериализации часто хочется плоский контейнер: DTO без методов, только поля. Наивный dataclass с __slots__ аллоцирует объект и хранит ссылки, protobuf-обертка — еще один слой. Решение: метакласс и __init_subclass__ Метакласс во время создания класса сам подменяет __slots__ и распластывает вложенные protobuf-сообщения в плоскую структуру:

class FlatMeta(type):
    def __new__(mcs, name, bases, namespace):
        proto = namespace.get('_PROTO')
        if proto:
            slots = tuple(field.name for field in proto.DESCRIPTOR.fields)
            def __init__(self, **kwargs):
                for name, val in kwargs.items():
                    object.__setattr__(self, name, val)
            namespace['__slots__'] = slots
            namespace['__init__'] = __init__
        return super().__new__(mcs, name, bases, namespace)

class UserDTO(metaclass=FlatMeta):
    _PROTO = UserProto

Теперь UserDTO — плоский объект с __slots__. Без лишних аллокаций при копировании. Почему это вывозит под high-flow * Нет __dict__ — объект занимает ровно размер полей плюс заголовок * Нет кэша валидации — все решается на этапе компиляции класса * Можно пилить напрямую в SerializeToString, без перегонки в словарь Типичная ошибка Использовать здесь dataclass с декоратором — он все равно создает __dict__ и добавляет лишний слой методов. Метакласс решает это на уровне создания класса. Практический совет Для production: такой подход годится только когда поля статичны и не требуют runtime-атрибутов. Для сложной валидации Pydantic все равно нужен. Но для чистого DTO это просто жир. Вывод: __init_subclass__ с метаклассами — инструмент, который выжимает наносекунды там, где каждый чих на счету, но требует строгой дисциплины в проектировании контрактов.