Ваш Python-сервис обрабатывает большое количество сетевых запросов. При профилировании видно, что он простаивает, ожидая I/O. Какой подход наиболее правильный для повышения производительности?
👾 — Увеличить количество потоков в ThreadPoolExecutor
👍 — Перейти на asyncio/uvloop и использовать асинхронные драйверы для работы с I/O
🥰 — Запускать каждый запрос в отдельном процессе через multiprocessing
⚡️ — Чаще вызывать gc.collect() для освобождения памяти
Библиотека задач по Python
👾 — Увеличить количество потоков в ThreadPoolExecutor
👍 — Перейти на asyncio/uvloop и использовать асинхронные драйверы для работы с I/O
🥰 — Запускать каждый запрос в отдельном процессе через multiprocessing
⚡️ — Чаще вызывать gc.collect() для освобождения памяти
Библиотека задач по Python
В Python-сервисе нужно обрабатывать большое количество CPU-bound задач (например, шифрование или обработку изображений). Вы используете ThreadPoolExecutor, но прироста производительности почти нет. Какой подход будет правильным?
👾 — Увеличить количество потоков в пуле до числа ядер × 10
👍 — Использовать ProcessPoolExecutor или multiprocessing, чтобы обойти GIL
🥰 — Переписать задачи на asyncio, чтобы они выполнялись конкурентно
⚡️ — Вставить вызовы gc.collect() внутри цикла, чтобы ускорить потоки
Библиотека задач по Python
👾 — Увеличить количество потоков в пуле до числа ядер × 10
👍 — Использовать ProcessPoolExecutor или multiprocessing, чтобы обойти GIL
🥰 — Переписать задачи на asyncio, чтобы они выполнялись конкурентно
⚡️ — Вставить вызовы gc.collect() внутри цикла, чтобы ускорить потоки
Библиотека задач по Python
👎1
Uvicorn/FastAPI-сервис использует asyncio и сторонние библиотеки, под капотом создающие потоки. Для CPU-bound задач вы подключили multiprocessing, но периодически ловите подвисания/краши при старте воркеров. Какой способ старта процессов корректнее выбрать?
👾 — fork — самый быстрый, поэтому всегда лучший выбор на Linux
👍 — spawn — запуск чистого интерпретатора, безопасен при наличии потоков/мьютексов у родителя
🥰 — forkserver — ничего не меняет относительно потоков, просто сложнее
⚡️ — Любой — способы эквивалентны на Linux
Библиотека задач по Python
👾 — fork — самый быстрый, поэтому всегда лучший выбор на Linux
👍 — spawn — запуск чистого интерпретатора, безопасен при наличии потоков/мьютексов у родителя
🥰 — forkserver — ничего не меняет относительно потоков, просто сложнее
⚡️ — Любой — способы эквивалентны на Linux
Библиотека задач по Python
В FastAPI вы кладёте correlation-id в contextvars.ContextVar в middleware. В логах он есть в хэндлерах, но иногда пропадает в коде, который выполняется через run_in_executor и в фоновых задачах asyncio.create_task(...). Что делать правильно?
👾 — Заменить ContextVar на глобальную переменную с мьютексом
👍 — Создавать фоновые задачи после установки ContextVar, а для run_in_executor запускать функцию через contextvars.copy_context().run(func, *args) для переноса контекста
🥰 — Перейти на threading.local(), он автоматически работает и в потоках, и в тасках
⚡️ — Включить uvloop — он сам перенесёт контекст в фоновые задачи и executor
Библиотека задач по Python
👾 — Заменить ContextVar на глобальную переменную с мьютексом
👍 — Создавать фоновые задачи после установки ContextVar, а для run_in_executor запускать функцию через contextvars.copy_context().run(func, *args) для переноса контекста
🥰 — Перейти на threading.local(), он автоматически работает и в потоках, и в тасках
⚡️ — Включить uvloop — он сам перенесёт контекст в фоновые задачи и executor
Библиотека задач по Python
Сервис создаёт миллионы однотипных небольших объектов (несколько фиксированных полей). Память растёт, GC давит. Что изменить в моделях, чтобы существенно снизить footprint и ускорить доступ к атрибутам?
👾 — Отключить GC через gc.disable()
👍 — Хранить данные в dict вместо объектов
🥰 — Использовать slots или @dataclass(slots=True) для моделей
⚡️ — Увеличить sys.setrecursionlimit()
Библиотека задач по Python
👾 — Отключить GC через gc.disable()
👍 — Хранить данные в dict вместо объектов
🥰 — Использовать slots или @dataclass(slots=True) для моделей
⚡️ — Увеличить sys.setrecursionlimit()
Библиотека задач по Python
В асинхронном сервисе (Python ≥ 3.11) нужно запустить группу независимых корутин и при первой ошибке гарантированно отменить остальные с корректной агрегацией исключений. Что выбрать?
👾 — asyncio.gather(*cors, return_exceptions=True)
👍 — asyncio.TaskGroup()
🥰 — loop.run_until_complete() по одной корутине в цикле
⚡️ — создать asyncio.create_task на каждую и игнорировать исключения в фоне
Библиотека задач по Python
👾 — asyncio.gather(*cors, return_exceptions=True)
👍 — asyncio.TaskGroup()
🥰 — loop.run_until_complete() по одной корутине в цикле
⚡️ — создать asyncio.create_task на каждую и игнорировать исключения в фоне
Библиотека задач по Python
Нужно запустить долгую корутину и дать вызывающему коду таймаут, но при истечении времени задача не должна отменяться, а продолжить выполняться в фоне. Что выбрать?
👾 — await asyncio.wait_for(op(), timeout=5)
👍 — t = asyncio.create_task(op()); await asyncio.wait_for(asyncio.shield(t), timeout=5)
🥰 — Запуск через ThreadPoolExecutor
⚡ — await asyncio.gather(op(), return_exceptions=True) с таймаутом
Библиотека задач по Python
👾 — await asyncio.wait_for(op(), timeout=5)
👍 — t = asyncio.create_task(op()); await asyncio.wait_for(asyncio.shield(t), timeout=5)
🥰 — Запуск через ThreadPoolExecutor
⚡ — await asyncio.gather(op(), return_exceptions=True) с таймаутом
Библиотека задач по Python
Нужно вызвать блокирующую функцию (нет async-аналога) из обработчика на asyncio, не блокируя event loop. Что выбрать?
👾 — Просто вызвать функцию напрямую в корутине
👍 — await asyncio.to_thread(func, *args, **kwargs)
🥰 — Обернуть функцию в async def и вызвать await func()
⚡ — Запустить функцию через time.sleep для «уступки» циклу
Библиотека задач по Python
👾 — Просто вызвать функцию напрямую в корутине
👍 — await asyncio.to_thread(func, *args, **kwargs)
🥰 — Обернуть функцию в async def и вызвать await func()
⚡ — Запустить функцию через time.sleep для «уступки» циклу
Библиотека задач по Python
👍1
Как в FastAPI (SQLAlchemy 2.0 async) корректно обеспечить «сеанс на запрос» с авто-rollback при исключении и гарантированным закрытием соединения?
👾 — Создать один глобальный AsyncSession и шарить его между хэндлерами
👍 — Использовать зависимость с yield
🥰 — Открывать/закрывать сессию вручную в каждом хэндлере try/except
⚡ — Класть сессию в app.state на старте приложения и переиспользовать
Библиотека задач по Python
👾 — Создать один глобальный AsyncSession и шарить его между хэндлерами
👍 — Использовать зависимость с yield
🥰 — Открывать/закрывать сессию вручную в каждом хэндлере try/except
⚡ — Класть сессию в app.state на старте приложения и переиспользовать
Библиотека задач по Python
В FastAPI сервисе при нагрузке появляются предупреждения Unclosed client session/connector is closed, растёт число открытых файловых дескрипторов. Вы используете aiohttp для внешних HTTP-запросов. Какой подход корректный?
👾 — Создавать aiohttp.ClientSession() на каждый запрос и не закрывать — пусть GC разберётся
👍 — Создать один ClientSession на старте приложения (lifespan), переиспользовать во всех хэндлерах и корректно закрыть на shutdown
🥰 — Заменить на синхронный requests внутри run_in_executor — тогда дескрипторы не утекут
⚡ — Для каждого запроса создавать новый TCPConnector(force_close=True) и сразу закрывать сессию
Библиотека задач по Python
👾 — Создавать aiohttp.ClientSession() на каждый запрос и не закрывать — пусть GC разберётся
👍 — Создать один ClientSession на старте приложения (lifespan), переиспользовать во всех хэндлерах и корректно закрыть на shutdown
🥰 — Заменить на синхронный requests внутри run_in_executor — тогда дескрипторы не утекут
⚡ — Для каждого запроса создавать новый TCPConnector(force_close=True) и сразу закрывать сессию
Библиотека задач по Python
Какое из следующих определений соответствует пакетам (packages) в Python?
Anonymous Poll
13%
Набор основных модулей
60%
Папка с модулями Python
20%
Набор файлов, содержащих определения и операторы Python
7%
Набор программ, использующих модули Python
Что из перечисленного не является ключевым словом в языке Python?
👾 — pass
👍 — eval
🥰 — assert
⚡️ — nonlocal
Библиотека задач по Python
👾 — pass
👍 — eval
🥰 — assert
⚡️ — nonlocal
Библиотека задач по Python
Нужно задать таймаут для блока асинхронных операций так, чтобы по истечении времени корректно произошло отмена вложенных await’ов и наверху ловился TimeoutError без «проглатывания» отмен. Что выбрать?
👾 — await asyncio.wait_for(coro, timeout=5)
👍 — async with asyncio.timeout(5): ...
🥰 — signal.alarm(5) в том же потоке
⚡️ — Оборачивать всё в собственный try/except CancelledError
Библиотека задач по Python
👾 — await asyncio.wait_for(coro, timeout=5)
👍 — async with asyncio.timeout(5): ...
🥰 — signal.alarm(5) в том же потоке
⚡️ — Оборачивать всё в собственный try/except CancelledError
Библиотека задач по Python
Какой будет вывод следующего фрагмента кода?
s = {1, 2, 3, 3, 2, 4, 5, 5}
print(s)
👾 — {1, 2, 3, 3, 2, 4, 5, 5}
👍 — {1, 2, 3, 4, 5}
🥰 — None
⚡️ — {1, 5}
Библиотека задач по Python
s = {1, 2, 3, 3, 2, 4, 5, 5}
print(s)
👾 — {1, 2, 3, 3, 2, 4, 5, 5}
👍 — {1, 2, 3, 4, 5}
🥰 — None
⚡️ — {1, 5}
Библиотека задач по Python
Как работает хэш-таблица?
Хэш-таблица в Python реализована в виде словаря (dict). Вот как это работает:
— Хэширование ключей: Когда вы добавляете пару ключ-значение в словарь, Python сначала вычисляет хэш-код ключа с помощью встроенной функции hash(). Хэш-код — это целое число, представляющее «отпечаток» ключа.
— Разрешение коллизий: Если два разных ключа имеют одинаковый хэш-код (коллизия), Python использует механизм разрешения коллизий для размещения значений в памяти. Одним из наиболее распространенных методов разрешения коллизий является метод цепочек, когда для каждого «ячейки» хэш-таблицы выделен список, в который добавляются все значения с одинаковыми хэш-кодами.
— Поиск значения: При поиске значения по ключу Python сначала вычисляет хэш-код ключа и затем использует его для определения соответствующей «ячейки» в хэш-таблице. Затем происходит поиск значения внутри этой «ячейки» (или цепочки).
Библиотека задач по Python
Хэш-таблица в Python реализована в виде словаря (dict). Вот как это работает:
— Хэширование ключей: Когда вы добавляете пару ключ-значение в словарь, Python сначала вычисляет хэш-код ключа с помощью встроенной функции hash(). Хэш-код — это целое число, представляющее «отпечаток» ключа.
— Разрешение коллизий: Если два разных ключа имеют одинаковый хэш-код (коллизия), Python использует механизм разрешения коллизий для размещения значений в памяти. Одним из наиболее распространенных методов разрешения коллизий является метод цепочек, когда для каждого «ячейки» хэш-таблицы выделен список, в который добавляются все значения с одинаковыми хэш-кодами.
— Поиск значения: При поиске значения по ключу Python сначала вычисляет хэш-код ключа и затем использует его для определения соответствующей «ячейки» в хэш-таблице. Затем происходит поиск значения внутри этой «ячейки» (или цепочки).
Библиотека задач по Python