⁣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% в месяц, иначе проще оставить как есть.

⁣Память под контролем: __slots__, pympler и гибридные модели

Про __slots__ пишут часто, но в реальных проектах с сотнями классов можно нарваться. Не всё так радужно.

Оверхед словаря и мифы

По умолчанию каждый экземпляр держит атрибуты в __dict__. Словарь с хеш-таблицей для миллионов объектов - это 50-70% оверхеда. __slots__ фиксирует список атрибутов на уровне класса и убирает __dict__ (если не добавить вручную). Экономия до 40-60%. Но в больших кодовых базах всплывают грабли:
- наследование: у родителя нет __slots__, у наследника есть - __dict__ все равно создается, выигрыша нет
- жесткость: нельзя динамически добавить атрибут - ломаются ORM, сериализаторы, половина библиотек
- дебаг: без __dict__ не увидишь текущие поля экземпляра в отладчике, приходится лезть в класс

Профилирование через pympler

Вместо гаданий - профилирование через pympler. Пример:

from pympler import asizeof

class WithSlots:
    __slots__ = ('x', 'y')
    def __init__(self):
        self.x = 10
        self.y = 20

class WithoutSlots:
    def __init__(self):
        self.x = 10
        self.y = 20

print(asizeof.asizeof(WithSlots()))  # ~72 байта
print(asizeof.asizeof(WithoutSlots()))  # ~184 байта

Цифры не врут: видно, где оверхед реально критичен, а где можно забить.

Гибридная модель: __slots__ + __dict__

Когда часть объектов динамическая (dataclass с optional полями), добавляем __dict__ в __slots__:

class FlexibleSlots:
    __slots__ = ('id', 'name', '__dict__')
    def __init__(self, id, name, **kwargs):
        self.id = id
        self.name = name
        for k, v in kwargs.items():
            setattr(self, k, v)

obj = FlexibleSlots(1, 'test', extra_field=42)
print(obj.extra_field)  # 42

Жесткие поля - без оверхеда словаря, динамические - через __dict__. Памяти чуть больше, но гибкость остается.

Что я обычно делаю:
- профилирую не на старте, а когда память реально начинает течь (pympler.muppy, pympler.summary)
- для тысяч объектов - ORM-модели, объекты игр - обязательно __slots__
- если dataclass с наследованием - @dataclass(slots=True) (Python 3.10+), но контролирую дерево наследования
- никогда не пихаю __slots__ в классы, которые подмешиваются через MRO или передаются в гибкие фреймворки

Вывод: __slots__ - мощный инструмент для сокращения памяти, но его применение требует профилирования и осознанного проектирования, особенно при наследовании и в гибридных сценариях с __dict__.