Сотни тысяч корутин лучше не запускать разом

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

Питоничная цель напоминает imap_unordered(): одновременно выполнять не больше limit операций и отдавать результаты по мере готовности. asyncio.Semaphore кажется очевидным ответом из Stack Overflow, но автор предупреждает: лучший вариант другой.

В разборе ограничения конкурентности сопоставлены gather(), Semaphore, as_completed(), Queue и wait(). Читайте, чтобы понять, почему автор выбрал wait() и как собрал асинхронный map_unordered() с поддержкой итерируемых объектов и исключений.

Внешние ключи Django могут оставить лишние индексы и заблокировать миграцию

В модели всё выглядит аккуратно: несколько ForeignKey, on_delete=PROTECT и unique_together. Но внешний ключ связывает две таблицы, поэтому обеспечить такое ограничение сложнее, чем уникальность или проверку значения. Явное лучше неявного, а неявного поведения здесь хватает.

Статья How to Get Foreign Keys Horribly Wrong разбирает, где появляются дублирующие индексы и как обнаружить блокирующую миграцию. Затем переходит к безопасному переносу внешнего ключа, обратимым операциям и конкурентному созданию индексов.

Перед следующим изменением схемы по ссылке стоит проверить ещё две вещи: когда нужен частичный индекс и в каком порядке выполнять миграционные операции.

Как найти хеш пароля в 37 ГБ меньше чем за миллисекунду

Офлайн-проверка пароля по списку утечек выглядит как обычный поиск, пока Pwned Passwords не распаковывается в текстовый файл на 37 ГБ. Минимальный вариант на Python работает, но оказывается слишком медленным.

Автор профилирует код, пробует пропускать части файла, применяет двоичный поиск, строит отдельный индекс и переводит его в двоичный формат. Вполне по дзену: сначала измерить, потом усложнять. Результат: поиск занимает меньше миллисекунды.

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

Как выбрать быстрый способ читать Excel из Python без сюрпризов с типами

Когда в Python нужно разобрать большой XLSX, привычный Pandas оказывается лишь одной из шести опций. Сравнивать их только по секундомеру мало: ещё важны сохранность типов и корректность значений.

Питоничность здесь начинается с контракта: каждая реализация возвращает Iterator[dict[str, object]], поэтому потребитель может обрабатывать строки по одной. Для теста взяли файл на 25 МБ с 500 тысячами строк, а время измеряли полным проходом без обработки данных.

В сравнении Haki Benita остались итоговые замеры для Pandas, Tablib, Openpyxl, LibreOffice, DuckDB и Calamine, а также разбор типов и корректности. Полезный ориентир перед тем, как ставить очередную зависимость ради одной таблицы.

Контекст запроса можно добавить во все логи без ручного extra

В ASGI-приложении запрос заканчивается не там, где начался: обработчик вызывает корутины из других слоёв, и каждая пишет собственные сообщения. Если добавить user_id и platform через extra только в обработчике, запись из вложенной функции останется без этих полей.

Явное лучше неявного, но собирать один и тот же контекст в каждом слое и вручную передавать его дальше — дорогая трактовка PEP 20.

В разборе распространения контекста логов показано, как общие поля попадают в сообщения из разных слоёв ASGI-приложения без ручной передачи по всей цепочке вызовов.

Сигналы Django подводят в критичных сценариях, но получателей можно выполнять надёжнее

Сигналы Django удобно разрывают зависимости между модулями. В обработке платежа этого мало: способ доставки сигналов остаётся подвержен неожиданным сбоям. Слабая связанность полезна, но надёжность от неё не появляется.

Альтернативный способ запускает получателей как фоновые задачи Django Tasks. В материале решение проверяют на процессе создания платежа и заказа, связывают с транзакциями базы данных, моделируют сбои и отдельно рассматривают атомарность.

В статье Reliable Django Signals остались тесты надёжных сигналов, ограничения подхода и направления дальнейшей работы. Стоит прочитать до того, как доверить сигналам этап процесса, который нельзя потерять из-за неожиданного сбоя.

Все дела подождут, потому что сегодня ваш день!

Поздравляем с Днем программиста и желаем меньше багов и больше фич, меньше созвонов и больше фокуса, меньше тасок с пометкой «срочно» и больше времени на отдых.

Забирайте подарок из нашей IT-коробки. Мы уверены, этот презент точно пригодится вам в работе: https://tprg.ru/la2S

Атрибуты модуля можно создавать только при обращении

Если часть API не нужна во время запуска, готовить её заранее необязательно. Начиная с Python 3.7, в модуле можно определить __getattr__: он обрабатывает обращение к имени, которого в модуле нет.

Для нужного имени функция возвращает динамически созданное значение, для остальных поднимает AttributeError. Так создание атрибута переносится на момент, когда он действительно понадобился, а запуск не ждёт этой работы. Лень здесь вполне питонична, пока ошибка для неизвестных имён остаётся явной.

В разборе атрибутов модуля есть минимальный пример с обычной функцией, динамическим атрибутом и AttributeError для неизвестного имени.

Когда isinstance() мало, TypeGuard может удивить

С isinstance() всё просто: mypy сам сужает объединение типов в каждой ветке. Сложнее, если объект приходится распознавать по содержимому, например проверять поля словаря. Тогда проверку выносят в отдельный предикат, а TypeGuard сообщает анализатору, какой тип прошёл условие.

На этом месте интуиция может подвести: автор несколько раз отказывался от TypeGuard и заканчивал комментарием # type: ignore. В разборе TypeGuard и TypeIs он начинает с TypedDict для Person и объясняет, почему TypeIs соответствует ожиданиям лучше.

Стоит прочитать перед следующим пользовательским предикатом: в статье осталось главное, почему TypeIs оказался тем интерфейсом, которого автор ожидал от TypeGuard.

Потоки ускоряют ввод-вывод, пока мелкая работа не упирается в процессор

Вы увеличиваете число рабочих потоков ThreadPoolExecutor, настраиваете пул соединений, а пропускная способность перестаёт расти. Даже короткий вычислительный участок повторяется в каждом задании и постепенно отбирает выигрыш от новых потоков.

Перейти на ProcessPoolExecutor можно, но тогда вход приходится разбивать на пачки, менять код и платить памятью за процессы. Автор собирает гибрид: несколько процессов, внутри каждого пул потоков, с привычным интерфейсом исполнителя задач.

В разборе ProcessThreadPoolExecutor остались возврат результатов через собственные объекты Future, гибель рабочего процесса и влияние свободнопоточного Python на всю конструкцию.

asyncio учат до первого исключения внутри задачи, а дальше начинается импровизация

Асинхронный код в проекте обычно дорастает до пары корутин и gather, а всё, что дальше, каждый достраивает по памяти: где ловить ошибку из задачи, упавшей молча, чем связывать корутины, кроме передачи результата в аргументах.

У Real Python есть разбор asyncio целиком, от корутин до выбора задач под них. Что в нём помимо привычного async/await:
• чем asyncio REPL (python -m asyncio) удобнее обёртки asyncio.run вокруг каждого эксперимента;
• как корутины стыкуются через asyncio.Queue, а не через цепочку await;
• что происходит с исключением, которое всплыло внутри задачи;
• в каких задачах asyncio выигрывает, а в каких только добавляет слой.

Открывать стоит ради последних двух пунктов: остальное вы, скорее всего, уже пишете руками.

#python

Как вызывать асинхронный Python из синхронного кода

Если оборачивать каждый вызов в asyncio.run(), Python каждый раз создаёт и закрывает цикл событий. Для одиночного запуска это нормально, но между вызовами не сохранятся привязанные к циклу ресурсы, например aiohttp.ClientSession с пулом соединений.

С Python 3.11 asyncio.Runner позволяет выполнять несколько корутин в одном цикле. Однако пока цикл занят, синхронная часть приложения ждёт. Следующий шаг: перенести цикл в отдельный поток и передавать ему корутины из обычных функций.

В подробном разборе автор собирает ThreadRunner, который запускает корутины и превращает асинхронные итераторы в обычные. Получается точечный мост для существующего синхронного приложения без переписывания всей цепочки в async def.

Как ускорить сериализацию в Django Rest Framework

В бенчмарке на Python 3.7, Django 2.1.1 и DRF 3.9.4 объект User сериализовали 5000 раз без обращений к БД. ModelSerializer потратил 12,8 с, а функция справилась за 0,034 с. Разница составила 377 раз.

Профилировщик показал цену автоматизации: ModelSerializer заново строил поля из метаданных модели и готовил проверки. Режим только для чтения сократил время до 7,4 с, а Serializer с явно объявленными полями уложился в 2,1 с. Явное оказалось не только лучше, но и быстрее.

Для нагруженного API отделите время запроса к БД от сериализации. Если тормозит последняя, сделайте выходные поля доступными только для чтения или замените ModelSerializer на Serializer; для предельной скорости сравните с обычной функцией. Код и профили cProfile есть в разборе производительности DRF.

Как тестировать HTTP-запросы в Python без хрупких моков

У асинхронного запроса через HTTPx есть четыре способа проверки. Можно вручную подменить AsyncClient.post, настроить асинхронный мок и проверить аргументы. С усложнением запроса растёт и эта обвязка.

Библиотека respx перехватывает запросы HTTPx и возвращает заготовленный ответ. Код короче, зато тест остаётся привязан к HTTPx. Более явный вариант в духе Python: передать клиент в функцию и на тесте заменить его объектом-заглушкой.

Для интеграционной проверки приложение Starlette играет роль тестового сервера, а AsyncClient обращается к нему вместо внешней сети. В статье есть код всех четырёх вариантов. Выбирайте respx для компактной подмены, заглушку вместо сторонней библиотеки для моков, тестовый сервер для проверки связки целиком.

Как собрать LRU-кеш с приоритетами и сроком жизни на Python

Обычный LRU-кеш удаляет давно не использовавшиеся записи. Здесь порядок сложнее: сначала просроченные элементы, потом записи с меньшим приоритетом, при равенстве — давно не запрашиваемые.

Автор начинает со словаря с операциями в среднем за O(1) и добавляет структуры для срока жизни, приоритета и истории обращений. Наивная очередь хранится в отсортированном списке: минимум легко прочитать, но вставка и удаление с начала требуют линейного времени. Затем очередь ускоряют через bisect, без куч и деревьев.

Разбор Адриана на death and gravity показывает, как согласовать несколько структур данных и проверить сроки через внедряемые часы. Это пример разработки от простого рабочего варианта к более быстрому только на стандартной библиотеке.

Как ждать корутины в asyncio

Два последовательных await не дают конкурентности: вторая корутина запустится после первой. Создание двух объектов корутин тоже не планирует их выполнение. Для этого нужны задачи, созданные до первого ожидания.

У каждого способа свои гарантии. gather() возвращает результаты в порядке аргументов. Если одна задача выбрасывает исключение, оно передаётся вызывающему коду сразу, а остальные задачи продолжают работать. as_completed() отдаёт результаты по мере готовности. wait() делит задачи на завершённые и ожидающие, а его тайм-аут сам ничего не отменяет.

В статье Waiting in asyncio автор советует начинать с TaskGroup, а для более гибкого управления выбирать wait(). Здесь «явное лучше неявного» работает буквально.

Как сделать параметризованные тесты pytest читаемыми

Обычный список кортежей в @pytest.mark.parametrize быстро усложняет чтение: значения приходится сопоставлять с аргументами по позиции, а идентификаторы сценариев хранятся в отдельном списке.

pytest.param помещает id рядом с данными. Тогда отдельный сценарий можно запустить командой pytest -k positive_x_axis. Через marks там же задаются xfail для ожидаемого сбоя и skipif для условного пропуска. Явное лучше неявного, особенно когда тест падает выборочно.

В статье Taming parametrize with pytest.param показан и следующий шаг: собрать именованные аргументы и ожидаемый результат в одном словаре. Такой вариант стоит примерить на тестах с длинной сигнатурой: у каждого сценария данные, имя и условия запуска оказываются рядом.

Как быстрее загружать грязные данные в PostgreSQL из Python

Автор берёт постраничный API с данными о пиве, превращает его в генератор и перед записью нормализует поля: извлекает объём из вложенного объекта, а месяц и год приводит к дате. Для замеров 325 записей дублируются 100 раз, получается набор из 32 500 строк.

Затем импорт сравнивается через построчные INSERT, executemany, execute_batch, execute_values и COPY. Для пакетных вариантов с итераторами меняется размер страницы, для COPY из строкового итератора — размер буфера. У каждого подхода измеряются время и пиковая память. Явное лучше неявного, особенно когда скорость не приходится угадывать.

В подробном разборе есть код генератора, преобразований, профилировщика и всех вариантов загрузки. Его удобно использовать как основу замера на реальных данных.

Наследование, композиция или функции: сравниваем на Executor в Python

Статья сравнивает три реализации гибридного исполнителя задач: наследование, композицию и функции. Он разносит работу по процессам, а внутри каждого запускает потоки, чтобы I/O-нагрузка использовала все ядра, когда упирается в CPU.

Наследник ProcessPoolExecutor переопределяет __init__, submit() и shutdown(), а map() и контекстный менеджер получает от родителя. При композиции map() приходится копировать для вызова нового submit(), а контекстный протокол писать заново. «Явное лучше неявного» внезапно означает больше кода.

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

Почему mock.patch не подменяет функцию в Python

После from os import listdir имя listdir привязано в модуле product. Последующая подмена os.listdir эту привязку не меняет, поэтому код продолжает вызывать product.listdir.

mock.patch работает как временное присваивание: заменяет имя по указанному пути, а после блока with возвращает исходное значение.

Практическое правило: подменяйте имя там, где его ищет тестируемый код. Здесь нужен mock.patch("product.listdir"). Он изменит поведение только product.py, не затрагивая другие модули. В статье Why your mock doesn’t work механизм разобран на двух способах импорта и схемах связей между именами.