Как вызывать асинхронный 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 для компактной подмены, заглушку вместо сторонней библиотеки для моков, тестовый сервер для проверки связки целиком.

Есть одна странная вещь в AI-индустрии.

Сегодня лучшая модель для кода — одна. Завтра выходит другая. Через месяц рынок снова меняется. Но большинство разработчиков всё ещё строят инфраструктуру так, будто им нужно выбрать одного AI-провайдера навсегда.

Polza.ai работает наоборот. Вы подключаете один API и получаете доступ к сотням моделей — OpenAI, Claude, Gemini, DeepSeek, Qwen и другим.

Нужна другая модель? Меняете её в запросе — и продолжаете работать. Можно сравнить провайдеров и цены и выбрать оптимальный вариант под конкретную задачу. А если один провайдер перестал отвечать, можно включить автомаршрутизацию — запрос автоматически уйдёт к другому.

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

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

Попробовать Polza.ai.

Как собрать 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 механизм разобран на двух способах импорта и схемах связей между именами.

Как устроен асинхронный веб-краулер на Python

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

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

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

В главе «A Web Crawler With asyncio Coroutines» из 500 Lines or Less код написан для Python 3.4. Читайте ради механики цикла событий; перед копированием сверяйте API с документацией.

Как изолировать Pydantic Settings от окружения в pytest

Если Settings создаётся при импорте пакета, он читает .env и переменные оболочки до запуска фикстуры pytest. Подмена os.environ запаздывает: значения уже сохранены в объекте, а тест зависит от окружения.

Решение автора: фикстура сохраняет model_copy() объекта, сбрасывает поля к значениям по умолчанию из model_fields и применяет точечные переопределения. Неизвестные настройки и неверные типы она отклоняет, после yield восстанавливает исходное состояние.

Подход пригодится, когда Settings уже живёт на уровне модуля. Явное снова лучше неявного. Код фикстуры и тесты со значениями по умолчанию и частичными переопределениями смотрите в разборе подмены Pydantic Settings.

Космический хакатон. Санкт-Петербург, Красноярск или онлайн из любой точки страны.

Старт 18 сентября.

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

Санкт-Петербург. К 2035 году над Землёй должен работать топливный узел, у которого заправляются корабли, и откуда он будет получать топливо — с Земли, с Луны или из резерва — пока не решил никто. Команде предстоит собрать схему поставок, посчитать резервы и решить, во что вкладываться сразу. Призовой фонд 1,2 млн ₽.

Красноярск. Спутник видит горящий лес раньше любого наземного поста, но вместе с пожаром ловит нагретые крыши, факелы на месторождениях и блики на воде. Команде предстоит научить сервис отличать настоящий очаг от шума, обвести гарь и выдать площадь в гектарах. Призовой фонд 600 тыс. ₽.

Лучшие забирают приз на площадке и билет в финал в Москве, 25–27 сентября, где разыграют ещё 2,4 млн рублей.

Команда 3–5 человек, недостающих можно найти на платформе.

Регистрация по ссылке — космохакатон.рф