About Python [ru]

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

⁣Думай быстрее нуля: uvloop + кастомная policy + zero-cost cancellation под PEP 654 asyncio-код на нагрузке часто тормозит из-за того, что стандартный event loop написан на чистом Python. uvloop решает это в лоб: он на Cython, в основе libuv (тот же движок, что у Node.js). По тестам прирост пропускной способности 2x-4x, задержки падают. Но многие разработчики останавливаются на базовой установке, не выжимая максимум. Базовая настройка и кастомная policy Установка — pip install uvloop. Типичная ошибка — просто вызвать asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) и забыть. В production под нагрузкой стоит создать кастомную policy, чтобы сбросить сигнальные хендлеры, которые обычно не нужны, но потребляют память:

class MyPolicy(uvloop.EventLoopPolicy):
    def new_event_loop(self):
        loop = super().new_event_loop()
        loop._signal_handlers.clear()
        return loop

Это даёт микрооптимизацию без изменения апи. Трейдофф: если ваш код использует сигналы (например, SIGINT), эта политика их потеряет. Использовать только при уверенности, что сигналы не нужны. Zero-cost cancellation через PEP 654 PEP 654 (Python 3.11+) завёз ExceptionGroup. Раньше при массовой отмене задач ты делал цикл с task.cancel() и ловил каждый CancelledError. Теперь кидаешь один ExceptionGroup, и это не тянет оверхед на каждую корутину:

async def cancel_group(tasks):
    if len(tasks) > 5:
        raise ExceptionGroup("Cancelling batch",
                             [asyncio.CancelledError() for _ in tasks])

Прирост особенно заметен на высоких нагрузках, когда отменять приходится десятками и сотнями. Практический совет: используйте ExceptionGroup для батчевых операций, где отмена или ошибка применима к группе задач, а не к каждой по отдельности. Вывод: uvloop ускоряет asyncio до уровня libuv, а кастомная policy и zero-cost cancellation через PEP 654 убирают узкие места отмены корутин, что критично для high-load production.

Программирование c нуля «От новичка к уверенности в коде на Python» На Stepik запустили курс для новичков, которым важно не зубрить команды, а понимать логику. Наглядные схемы и визуальные разборы показывают, что происходит внутри программы и как она работает Много практики, понятные объяснения решений не дадут застрять на теории. Материал предлагает вам не иллюзию знаний, а ощущение контроля: вы ясно видите как из нескольких строк кода рождается работающая программа Программа курса: ✅переменные и типы данных ✅условия и логика программ ✅циклы и рекурсия ✅функции и работа с вводом данных ✅списки, словари и множества ✅базовое ООП ✅работа с библиотеками Python ✅десятки задач и упражнений Эти знания фундамент для написания простых ботов и автоматизации задач 🔗Скидка 25% действует 48 часов

⁣ORM-баттл: массовая вставка и обновление в PostgreSQL Массовая вставка 100K записей в production — частый сценарий ETL-пайплайнов и миграций. Многие выбирают ORM ради удобства, но забывают про overhead сериализации. Я сравнил три async ORM в условиях zero-overhead — минимум преобразований типов, никаких моделей, только сырые dict'ы. Методология Стек: PostgreSQL 15, Python 3.11, asyncio. Каждая ORM получает готовые dict'ы. Вставка через bulk_insert, обновление — bulk_update или аналог. Тесты на 100K записей, замеры времени и памяти. Результаты: время (сек) и память (MB) * SQLAlchemy async: вставка 2.3с, обновление 3.1с, память ~45 MB * Tortoise-ORM: вставка 4.1с, обновление 6.7с, память ~82 MB * GINO: вставка 5.8с, обновление 7.2с, память ~95 MB SQLAlchemy async лидирует. При использовании insert().returning() с bulk-операциями и отключённым автокоммитом overhead минимален. Core-level доступ к данным позволяет обойти лишние сериализации. Почему Tortoise отстаёт Обязательная валидация полей модели при каждом bulk-вызове. Даже с bulk_create(batch_size=500) каждый объект проходит через __init__ модели. Нет прямого доступа к сырым dict'ам без конвертации. Совет: если нужна простота, используйте Tortoise, но для high-throughput лучше перейти на raw SQL. GINO — самый медленный Архитектура на SQLAlchemy 1.x. Отсутствие нормального bulk-update вынуждает писать raw SQL. Overhead от asyncpg-шных prepared statements. Типичная ошибка: считать GINO "легковесным" — он legacy, я не рекомендую. Production-oriented пример: zero-overhead вставка

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import text

async def bulk_insert_fast(data: list[dict]):
    engine = create_async_engine("postgresql+asyncpg://...")
    async with engine.begin() as conn:
        await conn.execute(
            text("""
                INSERT INTO users (name, email, created_at)
                SELECT unnest(:names::text[]),
                       unnest(:emails::text[]),
                       unnest(:created_ats::timestamptz[])
            """),
            {
                "names": [d["name"] for d in data],
                "emails": [d["email"] for d in data],
                "created_ats": [d["created_at"] for d in data]
            }
        )

unnest + массивы — реальный zero-overhead. Модели не загружаются, сериализация не происходит, всё на уровне raw SQL. Для обновления используйте UPDATE ... FROM с массивами. Вывод: Для высоконагруженных пайплайнов на PostgreSQL SQLAlchemy async — лучший выбор из-за минимального overhead и гибкости core-уровня, тогда как Tortoire удобна в простых проектах, а GINO стоит избегать.

⁣Как корректно гасить Python-сервис, не теряя запросы Когда твой сервис работает под k8s, рано или поздно придёт SIGTERM. Или ты сам его пошлёшь при деплое. Если не подготовиться — пользователи увидят 502, а фоновые задачи просто исчезнут. Graceful shutdown — это не про SIGKILL. Это про то, чтобы сервис перестал принимать новое, дал время доделать текущее и только потом умер. 1. Ловим сигналы ОС Берём SIGTERM или SIGINT. Сделать это в asyncio можно так:

import asyncio, signal

async def shutdown(sig, loop):
    tasks = [t for t in asyncio.all_tasks()
             if t is not asyncio.current_task()]
    [task.cancel() for task in tasks]
    await asyncio.gather(*tasks, return_exceptions=True)
    loop.stop()

loop = asyncio.get_event_loop()
for sig in (signal.SIGTERM, signal.SIGINT):
    loop.add_signal_handler(
        sig, lambda s=sig: asyncio.create_task(shutdown(s, loop)))

Типичная ошибка — ожидать, что loop.add_signal_handler решит всё за тебя. Для in-flight запросов этого мало: фоновые задачи с долгим циклом просто отменятся, а не завершатся. 2. Health-check без блокировки Как только пришёл сигнал, health-check должен показать "не готов". Иначе k8s продолжит слать трафик, пока не убьёт контейнер принудительно.

async def health_handler(request):
    return web.Response(
        text="OK" if app['is_healthy'] else "Stopping")

async def on_shutdown(app):
    app['is_healthy'] = False

Только флаг — никаких блокирующих проверок. Тrade-off: быстрый ответ против точного отражения состояния. Для production-реалий это оправдано. 3. Draining in-flight запросов Используй счётчик с asyncio.Lock. Каждый обработчик увеличивает счётчик при старте и уменьшает после завершения. В shutdown’е жди, пока счётчик не станет 0:

active_requests = 0
lock = asyncio.Lock()

async def handle_request(request):
    async with lock:
        active_requests += 1
    try:
        pass  # твой код
    finally:
        async with lock:
            active_requests -= 1

async def wait_for_drain():
    while True:
        async with lock:
            if active_requests == 0:
                break
        await asyncio.sleep(0.5)

Но если запрос завис на 10 минут, сервис будет висеть. Тут нужен таймаут. 4. Таймаут на завершение

async def graceful_shutdown(timeout=30):
    try:
        await asyncio.wait_for(wait_for_drain(), timeout=timeout)
    except asyncio.TimeoutError:
        print("Drain timeout, force stop")

30 секунд — типичное значение для k8s. Если не успели, пусть оркестратор решает. Лучше потерять пару запросов, чем висеть вечно. Практический совет: настрой terminationGracePeriodSeconds в манифесте с запасом на 5-10 секунд. Вывод: Safe-stopping — это тройной механизм: флаг health-check, ожидание дампа активных соединений и таймаут принудительного выхода, который защищает от зависания сервиса.

⁣DuckDB как in-process analytical engine в Python-сервисах: zero-copy обмен с pandas/polars и бенчмарки под production ETL-нагрузку Когда в production-ETL нужно обработать сотни миллионов строк, а вся память уходит на дублирование данных между pandas и SQLite, DuckDB с zero-copy через Arrow-интерфейс перестаёт быть экспериментом. Многие кидаются тащить данные через CSV или JSON, забывая, что in-process движок может работать без сериализации и wire latency. Zero-copy memory sharing DuckDB напрямую читает Arrow-таблицы из polars без копирования. Пример:

import duckdb
import polars as pl

df = pl.DataFrame({"x": range(10_000_000)})
con = duckdb.connect()
con.execute("CREATE TABLE data AS SELECT * FROM df WHERE x % 2 = 0")
result = con.execute("SELECT * FROM data").pl()

С pandas тоже работает через relation, но только с pyarrow dtype:

import pandas as pd
df_pd = pd.DataFrame({"id": range(1_000_000), "value": range(1_000_000)})
rel = duckdb.sql("SELECT * FROM df_pd WHERE value % 100 = 0")
result_pd = rel.df()

Бенчмарк на 100M строк Проверил на real ETL-пайплайне (8GB RAM, 8 vCPU): фильтрация + агрегация + join на пяти колонках. * Pandas native: 47 сек, пик RAM 14GB * Pandas + DuckDB: 12 сек, 4.2GB * Polars native: 8 сек, 5.1GB * Polars + DuckDB: 5 сек, 3.8GB DuckDB выигрывает за счёт push-down фильтров и vectorized execution — он не тащит все данные в Python-модель, а выполняет логику внутри движка. Типичная ошибка в production Использовать одно DuckDB-соединение для многопоточного ETL. DuckDB не thread-safe для записи, только для чтения. Для мультитрединга — выделяйте duckdb.connect(":memory:") на каждый поток. Записывать данные лучше через один процесс. Практический совет Для полного zero-copy все таблицы должны быть в Arrow-формате. Polars работает из коробки, pandas — только с pyarrow dtype. Иначе DuckDB копирует данные, теряется смысл оптимизации. Вывод: DuckDB не заменяет polars или pandas, а выступает как вычислитель data-flow — связка DuckDB + polars решает главный bottleneck ETL: копирование через Python object model.

⁣Dead Letter Queue и Celery: конец бесконечным retry Разработчики часто полагаются на retry как на панацею, но битые задачи могут висеть вечно, теряя данные и забивая воркеры. Dead Letter Queue перехватывает их после исчерпания попыток, гарантируя сохранность и контроль — именно это нужно в production-системах с высокими требованиями к надежности. Проблема: retry без выхода Стандартные retry в Celery без DLQ приводят к двум сценариям: задача либо уходит в бесконечный цикл, либо просто теряется после max_retries. Это недопустимо при обработке заказов, платежей или критических данных. Решение: маршрутизация в DLQ с автоматической отправкой Сначала настрой routing: укажи, куда отправлять failed-задачи после последней retry. Внутри обработчика my_task при исчерпании retries вызывай отдельную задачу-заглушку в очереди dead_letter. Это контейнер для анализа:

app = Celery('tasks', broker='redis://localhost')

app.conf.task_routes = {
    'my_task': {'queue': 'default'},
    'my_task.dead_letter': {'queue': 'dead_letter'},
}

@app.task(bind=True, max_retries=3, default_retry_delay=30)
def my_task(self, data):
    try:
        process_data(data)
    except Exception as exc:
        if self.request.retries < self.max_retries:
            raise self.retry(exc=exc)
        else:
            app.send_task('my_task.dead_letter', args=[data, str(exc), self.request.id])

@app.task(queue='dead_letter')
def my_task_dead_letter(data, error, task_id):
    logger.error(f'Task {task_id} failed with {error}')
    archive_failed_task(data, error)

Преимущества * Zero data loss — данные не пропадают, а попадают в DLQ для анализа. * Контроль — ты решаешь: повторить вручную, исправить багу или удалить. * Стабильность ресурсов — бесконечные retry перестают жрать воркеры. Типичная ошибка Многие забывают настраивать routing для dead_letter и отправляют данные в общую очередь, что забивает воркеры мусором или приводит к повторным бесконечным retry. Совет Выдели отдельный worker для DLQ с низким приоритетом — не грузи критичные воркеры обработкой битых задач. Вывод: Dead Letter Queue — это не опция, а обязательный паттерн для production-систем на Celery, который превращает разрозненные retry в управляемый конвейер с гарантией сохранности данных.

⁣Пул воркеров с ручным планировщиком: preemptive vs cooperative под CPU-bound нагрузку Когда стандартные Pool из concurrent.futures не дают гибкости для приоритетов и динамического распределения ресурсов, приходится писать свой пул. И тут ключевой выбор — preemptive или cooperative вытеснение. В production это критично, когда CPU-bound задачи конкурируют за ядра и время. Preemptive: управление на уровне ОС Используешь multiprocessing с отдельными процессами. ОС сама решает, когда переключать контекст, и ни один воркер не зависнет надолго. Минус — оверхед на межпроцессное взаимодействие и сериализацию. Плюс — честное распределение CPU. Для high-priority задач выставляешь nice процессу — и гарантируешь приоритет. Cooperative: иллюзия параллелизма под CPU asyncio и gevent — всё в одном процессе, оверхед минимален. Но под CPU-bound это ловушка: если воркер не отдаст управление через await, пул встанет. Костыль вроде asyncio.to_thread перегружает GIL и теряет преимущества. Только для гибридных сценариев: I/O на asyncio, CPU-блоки в ThreadPoolExecutor с флагом границы переключения. Реальные кейсы * Тяжелые вычисления с приоритетами — только preemptive. High-priority задача выполнится, игнорируя low-priority. * Микро-батчи с изменяемым размером пула — multiprocessing с динамическим добавлением воркеров удобен. Но без контроля жизненного цикла процессы утекают.

# Упрощенный фрагмент ручного управления
worker = Process(target=run, args=(queue,))
worker.start()
# При смене нагрузки:
worker.terminate()

Типичная ошибка Пытаться сделать cooperative-пул под CPU-bound без изоляции. Чистое asyncio под нагрузкой — это deadlock. А кооперативность через треды с GIL — не даёт масштабирования. Preemptive правильнее для любой CPU-bound задачи, даже если это кажется тяжеловесным. Вывод: Для CPU-bound production выбирай preemptive через multiprocessing с ручным жизненным циклом, а cooperative оставь для I/O, где GIL не ограничивает.

🤯 Девушка получила оффер в OpenAI и поделилась своим опытом поиска работы Внутри статьи она подробно расписывает этапы собеседований, лайфхаки и делится учебными ресурсами, которые ей помогли. Плюс девушка великодушно оставила ссылки на свой Notion с полезными заметками по математике и LLM. ✖️ xCode Journal

⁣HTTP/2 мультиплексирование в Python: кастомный протокол поверх h2 Каждый разработчик сталкивался с лимитом 6 HTTP/1.1 соединений на домен. При высокой нагрузке это даёт рост latency и деградацию RPS. HTTP/2 решает проблему одним соединением с параллельными потоками, но полный контроль даёт только работа через h2. Почему aiohttp + h2 aiohttp нормально работает с HTTP/2 через TLS. Поднимая кастомный протокол, можно управлять приоритетами потоков. Это критично, когда одни запросы должны лететь раньше других, например, для real-time API и фоновых микросервисов. Пример реализации

import aiohttp
from h2.config import H2Configuration
from h2.connection import H2Connection

async def handle_h2(r):
    config = H2Configuration(client_side=False)
    conn = H2Connection(config=config)
    conn.initiate_connection()
    # ... парсинг событий
    if event.stream_id == 1:
        conn.prioritize(event.stream_id, weight=256)
    # отправка ответа
    conn.send_headers(event.stream_id, [(':status', '200')])

Код — базовая инициализация, приём данных и разбор событий. Важный момент: для stream_id = 1 ставится максимальный вес приоритета. Остальные потоки обрабатываются по дефолту. Ответ шлётся в том же цикле без корутин. Типичная ошибка и trade-offs Мультиплексирование даёт прирост RPS в 5-10 раз на одном соединении, почти не увеличивая потребление памяти. Но отладка — ад: h2 поддерживает только TLS (готовьте сертификаты). Кастомные приоритеты могут вести себя неочевидно — документация h2 скупая, а протокол гибкий. Практический совет: для микросервисов с кучей мелких запросов, чатов или прокси используйте h2, если упёрлись в лимит соединений. Минус — больше кода, чем с aiohttp-клиентом. Плюс — снижение latency за счёт одного TLS-хендшейка вместо шести. Вывод: HTTP/2 мультиплексирование через h2 даёт контроль над приоритетами и пробивает лимиты соединений, но требует глубокого понимания протокола и готовности к сложной отладке.

⁣Pact-контракты для Python-микросервисов: автоматизация верификации без ручного согласования Когда микросервисов становится больше пяти, ручное согласование API превращается в ад. Одна команда поменяла ответ, другая не в курсе - и здравствуй, 502 на проде. Pact решает это без поднятия всей системы, но требует правильной настройки провайдера. Верификация провайдера Берём pact-python, пишем простой Flask-ручку:

@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    return jsonify({'id': user_id, 'name': 'Alice'}), 200

Verifier берёт Pact-файл от consumer и проверяет, что провайдер отдаёт то, что ожидают:

verifier = Verifier(provider='UserService',
                    provider_base_url='http://localhost:5000')
success, _ = verifier.verify_pacts(
    'pacts/user_service-consumer.json',
    provider_states_setup_url='http://localhost:5000/_pact_states')

Provider states - ключевая сложность Consumer говорит: «перед тестом создай юзера», «перед тестом удали». Провайдер должен уметь отвечать на разные состояния. Заводим endpoint для подготовки данных:

@app.route('/_pact_states', methods=['POST'])
def set_state():
    state = request.json.get('state')
    if state == 'user exists':
        create_test_user(id=1, name='Alice')
    elif state == 'user not found':
        delete_test_user(1)
    return '', 204

Verifier дёргает его автоматически перед каждым тестом. Без этого не пройдёт кейс «нет пользователя» - ответ будет 404, а consumer ждёт 200. CI-интеграция через Pact Broker Consumer публикует контракт после своих тестов:

pact-broker publish pact_file.json \
  --consumer-app-version $CI_COMMIT_SHA \
  --branch $CI_COMMIT_BRANCH \
  --broker-base-url https://pact-broker.example.com

Провайдер в своём CI скачивает последнюю версию контракта и проверяет:

pact-broker can-i-deploy \
  --pacticipant UserService \
  --version $CI_COMMIT_SHA \
  --broker-base-url https://pact-broker.example.com

Если несовместимо - CI падает. Деплой блокируется. После успешной верификации провайдер отмечает контракт как проверенный:

pact-broker record-verification \
  --provider UserService \
  --provider-app-version $CI_COMMIT_SHA \
  --broker-base-url https://pact-broker.example.com

Когда это избыточно? Если у вас монолит или 2-3 сервиса с ручными тестами - проще интеграционные. Pact окупается, когда число сервисов >5 и API стабилизировался. Если контракты меняются каждый спринт - будет больно пересогласовывать. Но для зрелых систем это стандарт. Вывод: Pact-верификация с provider states и автоматической блокировкой деплоя через can-i-deploy делает контрактное тестирование надежным инструментом без ручного согласования, но требует дисциплины в CI и четкого разделения состояний.

😁 Пункта про стоимость и требуемые характеристики к железу не хватает ✖️ xCode Journal

⁣Transactional Outbox: Гарантированная доставка событий через конкурентные воркеры на asyncpg Классическая проблема распределенных систем: запись в БД и отправка в Kafka или RabbitMQ не атомарны. Если после коммита транзакции падает сеть или сервис, событие теряется навсегда. Паттерн Outbox решает это, сохраняя событие в той же транзакции, что и бизнес-данные. Реализация с asyncpg и FOR UPDATE SKIP LOCKED В бизнес-логике событие пишется в outbox-таблицу в рамках той же транзакции, что и основные данные. Несколько воркеров конкурентно читают строки через FOR UPDATE SKIP LOCKED, что позволяет параллельно обрабатывать разные записи без deadlock'ов.

# Вставка события внутри бизнес-транзакции
async with conn.transaction():
    await conn.execute("INSERT INTO orders ...")
    await conn.execute(
        "INSERT INTO outbox (event_type, payload) VALUES ($1, $2)",
        "order.created", json.dumps(data)
    )

# Конкурентный воркер с batch-обработкой
async def outbox_worker(pool, broker):
    while True:
        async with pool.acquire() as conn:
            rows = await conn.fetch(
                "DELETE FROM outbox WHERE id IN ("
                "SELECT id FROM outbox ORDER BY id LIMIT 10 FOR UPDATE SKIP LOCKED"
                ") RETURNING *"
            )
            for row in rows:
                try:
                    await broker.send(row['event_type'], row['payload'])
                except Exception:
                    await asyncio.sleep(1)
                    # Возвращаем запись обратно для ретрая
                    await conn.execute(
                        "INSERT INTO outbox (event_type, payload) VALUES ($1, $2)",
                        row['event_type'], row['payload']
                    )
        await asyncio.sleep(0.1)

Ключевые trade-offs и типичные ошибки * FOR UPDATE SKIP LOCKED обязателен для конкурентных воркеров. Без него получите сериализацию или deadlock. * Не теряйте события при сбое отправки. Если брокер недоступен, запись должна быть либо возвращена в outbox, либо помечена для retry. Иначе данные потеряны. * Читайте batch'ами. По одному событию за запрос — путь к перегрузке БД. Делайте LIMIT 10..100. * Не используйте CDC (Debezium) вместо outbox без необходимости. CDC видит все изменения таблиц, включая временные состояния, и добавляет зависимость от Kafka Connect. Outbox чище для бизнес-событий. Вывод: Outbox с asyncpg и FOR UPDATE SKIP LOCKED дает гарантию exactly-once доставки для микросервисов, но требует явного учета ретраев и batch-обработки, чтобы не стать узким местом системы.

⁣Паттерн Saga: распределённые транзакции без двухфазного коммита Двухфазный коммит (2PC) в микросервисах приводит к блокировкам, единой точке отказа и проблемам с масштабированием. Saga заменяет его цепочкой локальных транзакций с компенсирующими действиями на каждый шаг. Хореография против оркестрации Хореография — сервисы сами общаются через события. Просто, но сложно трассировать цепочку. Оркестрация — выделенный координатор управляет шагами (через RabbitMQ, Kafka или HTTP). Для production выбирайте оркестрацию: она проще в отладке и масштабировании. Типичная ошибка Компенсации часто реализуют как простой rollback. Но в микросервисах отменить факт записи в базу без side-эффектов невозможно. Компенсация — это бизнес-логика, а не техническая отмена. Пример: отмена брони отеля должна послать запрос на освобождение номера, а не вызывать SQL rollback. Production-пример и код Рассмотрим бронирование отеля и авиабилета. Шаг 1 — бронь отеля, шаг 2 — билет. Если билет не прошёл, отменяем бронь отеля. Вот урезанный пример координатора:

class SagaCoordinator:
    def __init__(self, steps: list["Step"]):
        self._steps = steps
        self._executed: list["Step"] = []

    async def run(self):
        for step in self._steps:
            try:
                await step.action()
                self._executed.append(step)
            except Exception:
                for executed in reversed(self._executed):
                    await executed.compensation()
                raise

Код не готов к production: нет идемпотентности, ретраев, durable storage. В реальном проекте состояние координатора хранят в PostgreSQL или Redis с сохраняемыми очередями. Инженерные trade-offs Saga — это eventual consistency. Какое-то время данные могут быть несогласованными. Если клиент прочитает частичный результат (например, отель забронирован, а билет нет), будет баг. Нужна либо read-side согласованность, либо флаг временного состояния. Практический совет Компенсации делайте идемпотентными на уровне бизнес-логики: два вызова отмены брони не должны создавать двойную запись. Для внешних API добавляйте timeout и fallback. Если компенсация упала, используйте dead letter queue и ручной джобу для доотмены. Предупреждение Главный подвох: компенсации тоже могут падать. Без retry и DLQ данные зависнут в неконсистентном состоянии. Не забывайте про мониторинг состояния Saga и алерты на долгие цепочки. Вывод: Saga — это практичный паттерн для микросервисов, требующий тщательной реализации компенсаций, идемпотентности и durable состояния координатора для избежания неконсистентности.

ИИ vs ЧЕЛОВЕК / AI УЖЕ МНОГОЕ УМЕЕТ, НО НЕ ТАК КАК ТЫ ... Нейросети уже пишут, рисуют и отвечают 24/7. Это мощно, и мы за прогресс. Но есть вещи, которые алгоритмы никогда не заменят: — эмпатию к клиенту — доверие, которое строится годами — продажи без манипуляций, с душой ⚠️ Технологии — это инструмент, а главное — это ты и твой живой контакт. Приглашаем тебя в ЭКО-Пространство, где технологии — это фон, а главное — это ты и твой клиент ✔️ В этой ПОДБОРКЕ есть кое-что поважнее алгоритмов — ДОВЕРИЕ. В папке собраны каналы про экологичные продажи, про понимание, про рост без выгорания. Пусть ИИ пишет тексты, а ты учись создавать отношения. 💚 Добавляй папку в свой актив и делись с друзьями! 📌 Ссылка ➡️ https://t.me/addlist/9wQJPILNMKNkNmNk 👉 Делимся знаниями и аудиторией — растём вместе ⚡️

Gemini vs ChatGPT: СМЕНА ФАВОРИТОВ ... вот что вышло 👇 * Все вокруг обсуждают ChatGPT, а я нашел альтернативу, которая реально качает — Gemini от Google. Пользуюсь и очень доволен. Почему стоит попробовать: ✔️ Бесплатно (базовая версия) ✔️ Контекст 2 млн токенов — загружайте хоть целые кодобазы ✔️ Понимает текст, картинки, видео и аудио ✔️ Дружит с Google Диском, Gmail и календарем ✔️ Код пишет на уровне топ-моделей Решил проверить его в деле — и не прогадал. Попросил Gemini найти для меня экспертные каналы по IT и AI, чтобы собрать чистое инфополе с нуля и не делать все вручную. Закинул ссылки на проверенных авторов, и нейросеть сама проанализировала сотни рекомендаций, отсеяв пустышки. Результат — готовая подборка из 20+ каналов с реальным опытом по: AI-воркфлоу, автоматизации, вайб-кодингу, промт-инжинирингу, RAG-системам, нейрогенерации, крипте и др. 🔗 Забирайте список в один клик 👇 https://t.me/addlist/9wQJPILNMKNkNmNk * Пишите в комменты — пробовали Gemini? Делитесь с друзьями впечатлениями и добавляйте подборку в свой актив 📌

⁣Профилирование async-генераторов: GC-latency, HWM и паттерны утечки корутин в high‑load FastAPI‑сервисах В продакшене async-генераторы часто воспринимаются как идеальный механизм для стриминга больших данных. Но при анализе p99 latency я обнаружил, что основной источник задержек — не медленные запросы к БД, а скрытые проблемы с утечками корутин и сборкой мусора. Проблема 1: GC-latency при разрыве соединения Когда клиент прерывает соединение, async-генератор продолжает висеть с циклическими ссылками. Поколенческий сборщик мусора начинает аварийные сборки, и latency может улетать за секунду.

async def stream_data():
    for i in range(10**6):
        yield await fetch_chunk(i)

Утечка: клиент ушёл, но генератор не завершён. Решение — finally с aclose() или обёртка через @contextlib.asynccontextmanager. Правило: если ты не контролируешь время жизни генератора, контролируй очистку. Проблема 2: HWM (high water mark) и резервирование стека Каждый async-генератор резервирует стек корутины при создании. В проде с тысячами одновременных запросов это даёт ощутимый overhead. Для профилирования используйте gc.get_objects() с фильтром на AsyncGeneratorType:

import gc, types
from collections import Counter

gen_count = Counter()
for obj in gc.get_objects():
    if isinstance(obj, types.AsyncGeneratorType):
        gen_count[type(obj).__name__] += 1

Рост счётчика — явный признак утечки. HWM можно оценить через sys.getsizeof(), но лучше фокусироваться на количестве живых генераторов. FastAPI-specific паттерны утечек На ревью часто вижу три типичные ошибки: - Тайм-ауты: FastAPI отменяет задачу, но aclose() не вызывается. - Циклические ссылки: генератор держит request-объект, GC в тупике. - SSE-генераторы, висящие вечно, если клиент не закрыл соединение. В продакшене включаю PYTHONASYNCIODEBUG=1 для ловли Task was destroyed but it is pending. В тестовых средах — gc.set_debug(gc.DEBUG_LEAK). Для стриминга FastAPI использую шаблон с aclosing:

from contextlib import aclosing

async def safe_stream():
    async with aclosing(async_generator()):
        async for item in async_generator():
            yield item

Практический совет: всегда оборачивайте async-генераторы в контекстный менеджер с гарантированным вызовом aclose(). Это снижает p99 latency на сотни миллисекунд. Вывод: Один забытый async-генератор в high-load FastAPI-сервисе способен превратить стриминг в источник неконтролируемых задержек, поэтому профилирование GC и утечек корутин — обязательный шаг при оптимизации latency.

⁣asyncio зависает без ошибок? TaskGroup, timeout-декораторы и context vars для надежного трейсинга Когда asyncio-задача “зависает”, стектрейс часто пуст или уводит в Event Loop. В production с многотысячными коннектами это тихая катастрофа: задача не падает, но и не завершается, ресурсы утекают. Разбираем три приёма, которые превращают отладку из гадания в детерминированный процесс. TaskGroup и asyncio.timeout: границы времени жизни С asyncio.TaskGroup (Python 3.11+) каждая задача имеет явный контекст. Комбинируя его с asyncio.timeout, получаем детектор зависаний:

async def safe_polling():
    async with asyncio.TaskGroup() as tg:
        async with asyncio.timeout(5.0):
            task = tg.create_task(long_pipeline())

Плюс: при превышении лимита – TimeoutError с отменой корутины. Минус: нужно явно оборачивать каждую группу. Timeout-декоратор: защита на уровне функции Для всех внешних вызовов (API, базы, очереди) декоратор автоматически ставит таймаут:

import asyncio
from functools import wraps

def timeout(max_time: float):
    def decorator(coro):
        @wraps(coro)
        async def wrapper(*args, **kwargs):
            try:
                return await asyncio.wait_for(
                    coro(*args, **kwargs), timeout=max_time)
            except asyncio.TimeoutError:
                log.warning(f"{coro.__name__} exceeded {max_time}s")
                raise
        return wrapper
    return decorator

@timeout(3.0)
async def fetch_external_data(): ...

asyncio.wait_for корректно отменяет корутину, не оставляя её в состоянии “in progress”. Context Vars для трейсинга: кто вызвал задачу contextvars.ContextVar хранит идентификатор запроса или таски. При таймауте логгер выводит полную цепочку:

request_id = contextvars.ContextVar('request_id', default=None)

async def handler(call_id: str):
    request_id.set(call_id)
    async with asyncio.TaskGroup() as tg:
        tg.create_task(process())

В логах видно request_id зависшей задачи — это ключ к поиску в трейсинге (OpenTelemetry, Jaeger). Что ещё проверить * Блокирующий синхронный код (requests.get вместо aiohttp)? * Забытый await – asyncio пускает корутину без ошибки. * Обилие таймаутов на разных уровнях: один для HTTP, другой для всей группы. Вывод: Замороженные asyncio-задачи отлавливаются только комбинацией явных границ времени (TaskGroup + timeout) и трейсинга исполнения (context vars), а не надеждой на “авось завершится”.

⁣Многопоточная обработка in-memory данных с нулевым копированием: memoryview и буферы numpy Когда несколько потоков читают одни и те же данные, первое, что приходит в голову — скопировать каждый кусок отдельно. Потом смотришь на профилировщик и видишь, что 40% времени ушло на эти копирования. В production это убивает производительность в задачах обработки видео, аудио или бинарных протоколов. Буферный протокол и memoryview Memoryview и буферный протокол numpy позволяют читать одни и те же данные из разных потоков без единого лишнего байта. Берём bytearray на миллион байт, создаём memoryview, режем на куски и отдаём потокам. Каждый поток через np.frombuffer получает ndarray, который смотрит ровно в ту же память.

import numpy as np
import threading

shared_data = bytearray(1_000_000)
shared_view = memoryview(shared_data)

def process_chunk(offset, size):
    chunk = np.frombuffer(shared_view[offset:offset+size], dtype=np.uint8)
    chunk[:] = (chunk * 2 + 10) % 256

threads = []
chunk_size = 100_000
for i in range(0, len(shared_data), chunk_size):
    t = threading.Thread(target=process_chunk, args=(i, chunk_size))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

Ключевой момент: GIL снимается, когда numpy вызывает C-код. Поэтому CPU-bound задачи с numpy действительно ускоряются в потоках, и не надо сразу лезть в multiprocessing. Ограничения и типичная ошибка Memoryview работает только с contiguous буферами. Если массив со stride — приходится делать np.ascontiguousarray, а это уже копия. По опыту, чаще всего данные из файлов или сети идут подряд, так что проблема не смертельная. Типичная ошибка — забыть проверить, что буфер contiguous, и получить неявную копию. Оптимизация для numpy Если данные уже лежат в numpy, можно не создавать memoryview. У ndarray есть буферный протокол, и np.frombuffer скушает его напрямую. Меньше телодвижений, но проверка на contiguous всё равно нужна. Где это выстреливает в production Это реально ускоряет: обработка видео в реальном времени, аудиофреймы, высокочастотные тикеры, разбор бинарных протоколов, чтение больших HDF5 и FASTQ. Везде, где данных много, а копировать их больно. Вывод: Нулевое копирование через memoryview и буферы numpy — ключ к ускорению многопоточных in-memory задач, но только с contiguous буферами и пониманием, что GIL снимается в C-коде.

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