⁣Профилирование утечек памяти в 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 — глобальные забытые объекты и циклические ссылки, и без них вы будете сидеть с логами и гадать, где память утекает.

Как я устал писать парсер под каждый прайс и сделал из этого библиотеку

На проекте десятки прайсингов на топливо: один вендор шлёт CSV, другой Excel, третий JSON на вебхук. Данные одни, но колонка цены везде называется по-своему, даты в трёх форматах, единицы то литры, то галлоны, а половина нужных полей отсутствует. Под каждый источник жил отдельный парсер на сотню строк if-else. Сначала их было три, потом восемь, потом количество перестали считать. Парсеры ломались молча: вендор тихо переименовывал колонку.

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

Источник

⁣Асинхронные краны сообщений через 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 на кластерных нагрузках.

DTO, schema, model, entity: почему в коде всё называется User

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

Читать далее

⁣Микрооптимизация сериализации в 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__ и кастомными кодеками дают реальный прирост производительности, но требуют строгого контроля над форматом данных и готовности к увеличению сложности поддержки.

⁣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-сервисах.

Новинка: «Инженерия данных. Паттерны проектирования»

Издательство «O’Reilly» выпустило русское издание книги «Data Engineering Design Patterns» Бартоша Конечны под названием «Инженерия данных. Паттерны проектирования». Книга вышла в конце июня 2026 года. Автор предлагает выделить в дисциплине инженерии данных универсальные шаблоны проектирования типичных решений, аналогичные тем, что описаны в книге «Design Patterns» «Банды четырёх» середины 1990-х. Ранее Бартош Конечны опубликовал статью-перевод, в которой обосновывал готовящуюся книгу и очерчивал её тематическое поле.

Читать далее

⁣Lock‑Free кэш на C11 atomics через Python – когда GIL не помогает, а блокировки давят

В многопроцессной архитектуре (например, между воркерами Gunicorn или Celery) обычные Lock из threading или multiprocessing становятся узким местом из‑за контекстных переключений. Lock‑free структуры на атомарных операциях C11 обходят это. Python позволяет дотянуться до них через ctypes и _multiprocessing.sharedctypes, работая напрямую с разделяемой памятью без GIL. Частая ошибка: разработчики пишут свой lock‑free код, не учитывая memory ordering и ABA‑проблему.

Как собирается атомарный кэш

Берёшь RawValue и RawArray из _multiprocessing.sharedctypes, выделяешь разделяемую память для флагов, ключей и значений. Для синхронизации используешь C11 __sync_bool_compare_and_swap (CAS) через ctypes.CFUNCTYPE – он вызывает atomic-инструкцию на уровне процессора.

from ctypes import c_uint64, c_bool, CFUNCTYPE, POINTER, byref
from _multiprocessing.sharedctypes import RawValue, RawArray

class LockFreeCache:
    def __init__(self, capacity=256):
        self.capacity = capacity
        self.keys = RawArray(c_uint64, capacity)
        self.values = RawArray(c_uint64, capacity)
        self.flags = RawArray(c_bool, capacity)
        self._load_cas()

    def _load_cas(self):
        libc = ctypes.CDLL(None)
        self.cas = CFUNCTYPE(c_bool, POINTER(c_bool), c_bool, c_bool)(
            ('__sync_bool_compare_and_swap', libc)
        )

    def set(self, key, value):
        idx = hash(key) % self.capacity
        while True:
            old_flag = c_bool(False)
            if self.cas(byref(self.flags[idx]), old_flag, c_bool(True)):
                self.keys[idx] = key
                self.values[idx] = value
                self.flags[idx] = c_bool(False)
                return True

Production‑ориентированный пример

Используй такой кэш на hot‑path, где простая блокичка через multiprocessing.Lock даёт ощутимый оверхед. Например, подсчёт запросов или кэширование результатов между воркерами в асинхронном HTTP‑обработчике. Для прода обязательно добавь управление коллизиями (хэш‑таблица с open addressing), TTL и видимость через memory fence (GCC __sync_synchronize).

Trade‑offs и типичные ошибки

Плюсы: без блокировок, масштабируется на многоядерных системах, GIL не мешает. Минусы: ABA‑проблема (например, чтение старой записи после перезаписи), платформозависимость (не на всех архитектурах есть CAS), риск data race при неправильном memory ordering. Предупреждение: не используй этот подход для сложных структур – lock‑free очередь или счётчик проще сделать через multiprocessing.Value с блокировкой.

Вывод: Lock‑free кэш на атомарных операциях оправдан только на узком hot‑path, где блокировка реально давит – в остальных случаях простой Lock надёжнее и читаемее.

Геостатистика в QGIS без SAGA: кригинг на чистом NumPy

Автор статьи делится опытом создания софта для горно-геологических служб калийных рудников. Геологи и маркшейдеры ежедневно превращают тысячи скважинных проб в карты: отметки кровли пласта, содержания KCl, мощности, газоопасность. Классический инструмент для этого — кригинг, который в QGIS формально есть через SAGA, GRASS, Smart-Map и связки со SciPy. Однако каждый из этих вариантов чем-то не устраивал, и год назад автор начал писать свой плагин.

Сейчас плагин Isoliner включает 24 инструмента в официальном репозитории plugins.qgis.org: кригинг четырёх видов, вариограммный анализ, кросс-валидация с отчётами, изолинии с контурными полигонами, геологические разрезы и собственный 3D-просмотр. Вычислительное ядро построено на чистом NumPy без внешних зависимостей.

В статье также объясняется, зачем понадобился ещё один кригинг, как выглядит система кригинга в двадцати строках NumPy, что такое вариограмма на пальцах и почему абсолютные единицы силла — главные грабли для новичков.

Читать далее на Хабре

⁣Сборка колл-стеков для профилирования CPU в production через sys.setprofile и signal.setitimer без внешних трейсеров

Когда прожорливый код угробит CPU в продакшене, а ставить py-spy или Pyroscope нельзя из-за политики безопасности или ограничений архитектуры, поможет связка стандартной библиотеки. Разработчики часто забывают, что семплер можно собрать за час без единой внешней зависимости, но допускают ошибки с потоками и задержками.

Идея и реализация
signal.setitimer запускает обработчик через заданные интервалы (например, 50-100 мс). Внутри него sys._current_frames() делает мгновенный снапшот стеков всех потоков. Ключевой нюанс: не используй traceback.format_stack() — это тормозит. Только сухие f_code.co_name и f_lineno.

Production-ready пример для мониторинга CPU-горячих точек

import signal
import sys
import threading
import time

stacks_buffer = []
BUFFER_LOCK = threading.Lock()

def collect_stack(signum, frame):
    try:
        snapshot = {}
        for tid, t_frame in sys._current_frames().items():
            stack = []
            while t_frame:
                stack.append(f"{t_frame.f_code.co_name}:{t_frame.f_lineno}")
                t_frame = t_frame.f_back
            snapshot[tid] = stack[:15]  # Ограничиваем глубину
        with BUFFER_LOCK:
            stacks_buffer.append((time.time(), snapshot))
    except Exception:
        pass  # Тихий сброс ошибок

signal.signal(signal.SIGALRM, collect_stack)
signal.setitimer(signal.ITIMER_REAL, 0.05, 0.05)  # 50 ms

# Далее обычный код приложения...
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    signal.setitimer(signal.ITIMER_REAL, 0, 0)
    for ts, snap in stacks_buffer[:5]:
        print(ts, snap)

Типичные ошибки и trade-offs
* Сигнал SIGALRM работает только в главном потоке — для многопоточности нужен ручной signal.signal и отдельный поток-коллектор для записи в буфер (иначе потеря данных из-за рекурсии сигналов).
* Минимальная частота — 50 мс; ниже уже жрёт CPU сам семплер. Для asyncio проще использовать asyncio.Task.get_stack() из цикла событий.
* C-расширения (например, numpy или lxml) не покажут стек — тут без py-spy или perf не обойтись.

Практический совет
Записывай в кольцевой буфер с фиксированной ёмкостью (например, 1000 сэмплов) и асинхронно сбрасывай на диск в фоновом потоке — это thread-safe и не фризит основной код.

Вывод: Связка sys.setprofile + signal.setitimer даёт дешёвый семплинг стека для поиска CPU-горячих точек в продакшене без внешних зависимостей, но требует аккуратности с потоками, частотой и обработкой C-расширений.

⁣Типизированные TypeVar и ParamSpec для generic-коллбэков: статическая проверка обработчиков событий без Protocol и лишних абстракций

Типизация коллбэков — вечная боль production-кода, особенно при реализации диспетчеров событий или callback-based middleware. Частая ошибка: использовать Callable[..., Any], теряя сигнатуру и получая TypeError на проде.

Проблема

def register(handler: Callable[..., Any]):
    ...

Здесь анализатор не проверяет аргументы. В production это приводит к ошибкам при первом несовпадении типов — например, когда обработчик ожидает (int, str), а передается (float, User).

Решение

from typing import TypeVar, ParamSpec, Callable

P = ParamSpec("P")
T = TypeVar("T")

def register(
    handler: Callable[P, T],
    *args: P.args,
    **kwargs: P.kwargs,
) -> T:
    return handler(*args, **kwargs)

Статический анализатор теперь видит точную сигнатуру коллбэка. Никаких сюрпризов.

Реальный пример: диспетчер событий

class EventDispatcher:
    def on(self, event: str) -> Callable[[Callable[P, T]], Callable[P, T]]:
        def wrapper(handler: Callable[P, T]) -> Callable[P, T]:
            self.handlers[event] = handler
            return handler
        return wrapper

dispatcher = EventDispatcher()

@dispatcher.on("user_login")
def handle_login(user_id: int, timestamp: float) -> str:
    return f"User {user_id} logged in at {timestamp}"

# Правильно
result = dispatcher.handlers["user_login"](42, 1689000000.0)

# Ошибка типов: expected float, got str
result = dispatcher.handlers["user_login"](42, "bad")

Почему это лучше Protocol?

* Меньше кода — не надо объявлять абстрактные классы с __call__
* Точность — сохраняются имена параметров, порядок, типы и возвращаемое значение
* Универсальность — один декоратор работает с любыми сигнатурами
* Совместимость — работает с callback-based либами вроде functools или asyncio

Когда Protocol все же нужен?

Если коллбэк должен иметь атрибуты (например, handler.priority = 5) или наследовать несколько абстракций. По опыту — в 80% случаев это избыточно.

Вывод: ParamSpec + TypeVar дают чистую и проверяемую типизацию generic-коллбэков без лишних сущностей — замените Callable[..., Any] на точные сигнатуры на ревью.

⁣Graceful Shutdown в asyncio: как не потерять данные при SIGTERM в production

Стандартный signal.signal блокирует event loop, превращая асинхронное приложение в синхронный ступор. Решение - loop.add_signal_handler(), но без pipe вы рискуете утечкой ресурсов: воркеры не успеют закрыть соединения или снять блокировки.

Проблема: почему просто cancel() не работает
Сигналы обрабатываются в том же потоке, что и цикл событий. Если в обработчике сразу отменять корутины, вы получите состояние гонки: одни задачи завершатся, другие - нет. Ресурсы утекут, соединения повиснут. Пример частой ошибки:

def handler():
    for task in asyncio.all_tasks():
        task.cancel()  # Блокировка в синхронном контексте

Решение: pipe как мост между синхронным и асинхронным миром
Используйте os.pipe() для безусловного пробуждения цикла. Обработчик сигнала только пишет байт в pipe - никаких корутин или блокировок. Через add_reader асинхронный контекст подхватывает запись и выполняет graceful shutdown:

class GracefulShutdown:
    def __init__(self):
        self.loop = asyncio.get_event_loop()
        self.r_fd, self.w_fd = os.pipe()
        self.loop.add_reader(self.r_fd, self._handle_stop)
        for sig in (signal.SIGTERM, signal.SIGINT):
            self.loop.add_signal_handler(sig, self._signal_handler)

    def _signal_handler(self):
        os.write(self.w_fd, b'\x00')  # Только запись

Критичное правило: никакой логики в _signal_handler
Запись в pipe - единственная операция. Закрытие дескрипторов, отмена задач, asyncio.all_tasks() - все это делается в асинхронном _handle_stop, где task.cancel() пробрасывает CancelledError в воркеры, давая шанс выполнить cleanup. Если нарушить это правило в многопоточном run_in_executor, словите блокировку цикла.

Trade-off: pipe vs call_soon_threadsafe
loop.call_soon_threadsafe тоже работает, но pipe надежнее в сценариях с воркерами в других потоках: он гарантированно будит именно тот цикл, который слушает. Без pipe вы рискуете, что сигнал пропустит цикл, занятый долгим await.

Вывод: Pipe + add_signal_handler превращает SIGTERM из источника утечек в управляемое завершение, где каждый воркер закрывает ресурсы через CancelledError.

Какой кригинг выбрать: простой, ординарный, с трендом, блочный, индикаторный

Мы создаем софт для горно-геологических служб калийных рудников, и после первой статьи про кригинг на чистом NumPy самый частый вопрос звучал одинаково: «Хорошо, а какой именно кригинг брать?» Вопрос правильный: под словом «кригинг» живёт целое семейство методов, и выбор между ними влияет на результат сильнее, чем тонкая настройка вариограммы.

В плагине Isoliner их пять - простой, ординарный, с полиномиальным трендом, блочный и индикаторный, - и каждый существует не для галочки, а под конкретный класс геологических задач. Все пять видов кригинга решают систему: оценка в точке — взвешенная сумма соседних скважин, веса — решение системы уравнений с ковариациями из вариограммы. Различаются они тем, что считается неизвестным про среднее поле и что именно оценивается — точка, блок или вероятность. Простой кригинг предполагает, что среднее значение поля известно заранее и постоянно по площади, и недобор веса соседей компенсируется этим средним.

Читать далее

⁣Zero-Copy Обёртки Над Буферами: Парсинг Бинарных Протоколов Через __buffer__ и __release_buffer__

Каждый раз, когда дамп сырого memoryview из сокета копируется в bytes для парсинга struct, теряется время — на production с десятками тысяч пакетов в секунду это становится узким местом. В Python 3.12+ протокол буфера позволяет реализовать zero-copy классы-обёртки без копирования данных.

Зачем нужны типизированные обёртки
Сырой memoryview — это всего лишь как бы массив байтов, где нужно помнить смещения и форматы. Класс с полями даёт типобезопасность и читаемость, при этом внутри работает с тем же буфером. Например, для фиксированного заголовка пакета: 4 байта ID, 2 байта длины, 2 байта флагов.

import struct

class PacketHeader:
    def __init__(self, buffer):
        self._buffer = buffer

    def __buffer__(self, flags):
        return self._buffer.__buffer__(flags)

    def __release_buffer__(self, buffer):
        pass

    @property
    def packet_id(self):
        return struct.unpack_from('<I', self._buffer, 0)[0]

    @property
    def length(self):
        return struct.unpack_from('<H', self._buffer, 4)[0]

Zero-copy парсинг в действии
Вместо копирования через bytes(data) используем MSG_PEEK и передаём memoryview напрямую:

raw_data = memoryview(sock.recv(8, socket.MSG_PEEK))
header = PacketHeader(raw_data)
print(header.packet_id, header.length)

Нет ни одного лишнего копирования. Это особенно критично для high-throughput протоколов в связке с asyncio, где каждый микросекунда на аллокацию — потеря.

Типичная ошибка и trade-offs
Управление временем жизни буфера — это головная боль. Если исходный memoryview будет уничтожен раньше обёртки, вы получите segfault или мусорные данные. Всегда явно контролируйте scope: держите ссылку на исходный буфер до завершения парсинга.

Вывод: Протокол __buffer__ и __release_buffer__ дают zero-copy доступ к бинарным данным без копирования, но требуют аккуратного менеджмента памяти и не заменяют быстрый struct.unpack для разовых задач.

⁣Zero-downtime миграции схем в SQLAlchemy 2.0: партиционирование и transactional DDL в Alembic

Когда таблица переваливает за 100 миллионов строк, обычная миграция через ALTER TABLE превращается в план на выходные с даунтаймом. Если повезёт.

Почему transactional DDL критичен
PostgreSQL умеет выполнять DDL внутри транзакций — это transactional DDL. Alembic это поддерживает, но при партиционировании важно не забыть про autocommit_block(). Без него некоторые операции (создание партиций) могут упасть вне транзакции, и сервер уйдёт в простой.

Expand-contract стратегия
Вот как выглядит zero-downtime подход:

Фаза 1 (Expand) — создаём новую партиционированную таблицу параллельно старой. Не трогаем существующую схему.

Фаза 2 (Migrate) — организуем двойную запись. Апдейты пишутся и в старую, и в новую структуру. Через триггеры или приложение.

Фаза 3 (Contract) — переключаем чтение на новую схему. Удаляем старую таблицу, когда убедились, что всё работает.

В коде Alembic это выглядит примерно так:

def upgrade():
    with op.get_context().autocommit_block():
        op.execute("""
            CREATE TABLE orders_new (LIKE orders INCLUDING ALL)
            PARTITION BY RANGE (created_at)
        """)
    # Дальше — батчевая вставка данных с паузами

Главные грабли
* Deferred constraints. Если есть внешние ключи, их лучше отключать на время миграции, иначе потом не запихнете данные.
* Batch processing. Не пытайтесь перелить 100M строк одним INSERT. Делайте чанками по 1000 записей с time.sleep(0.1). Иначе транзакция отожмёт всё.
* Versioned migrations. Сохраняйте старые партиции хотя бы неделю после переключения. На проде всегда вылезает "ой, а мы забыли перенести поле".

Production-oriented пример
Создаём теневую таблицу, льём данные батчами, потом атомарно переименовываем:

def upgrade():
    op.create_table('orders_shadow',
        sa.Column('id', sa.Integer, primary_key=False),
        postgresql_partition_by='RANGE (created_at)'
    )
    connection = op.get_bind()
    total = connection.execute("SELECT count(*) FROM orders").scalar()
    for offset in range(0, total, 1000):
        connection.execute(f"""
            INSERT INTO orders_shadow 
            SELECT * FROM orders 
            ORDER BY id LIMIT 1000 OFFSET {offset}
        """)
        time.sleep(0.1)
    connection.execute("""
        ALTER TABLE orders RENAME TO orders_old;
        ALTER TABLE orders_shadow RENAME TO orders;
    """)

Предостережения, которые обычно игнорируют
* Тестируйте на клоне прода. Не на тестовой базе с 10 строками.
* Используйте --sql флаг Alembic, чтобы посмотреть, что он сгенерирует, и руками проверить план.
* Мониторьте log_statement и deadlocks. Партиционирование не спасает от блокировок при массовом INSERT.
* Имейте rollback-скрипт для каждой фазы. Если что-то пошло на этапе переименования, откат должен быть за секунды.

Вывод: Партиционирование через transactional DDL — это не магия, а расчёт, который окупается, когда таблицы растут на 10% в месяц, иначе проще оставить как есть.