Как реализовать корректный graceful shutdown в asyncio-сервисе без потери данных?
Ловите SIGTERM/SIGINT (Unix — loop.add_signal_handler), ставьте общий флаг/Event, отменяйте воркеры через TaskGroup/cancel(), ждите их в try/except CancelledError/finally. Давайте тайм-аут на дренаж очередей и завершение in-flight операций, корректно закрывайте пулы (БД/HTTP). В ASGI используйте lifespan-хуки (startup/shutdown).
Библиотека собеса по Python
Библиотека собеса по Python
Как находить и устранять утечки памяти в долгоживущем сервисе?
Включите tracemalloc и снимайте снэпшоты до/после нагрузки, ищите рост по трассам; проверяйте циклы gc.get_objects() и gc.garbage, избегайте __del__ на участниках циклов, используйте weakref для обратных ссылок. Ограничивайте кэши (lru_cache(maxsize=…)), следите за «вечно живущими» задачами asyncio, закройте пулы/соединения и проверяйте, что обработчики событий не держат замыкания на большие объекты.
Библиотека собеса по Python
Библиотека собеса по Python
Как спроектировать контекстные менеджеры так, чтобы ресурсы гарантированно освобождались даже при отмене/исключениях (sync/async)?
Реализуйте __enter__/__exit__ или __aenter__/__aexit__, делайте освобождение идемпотентным и безопасным к повторному вызову, в __exit__ не глотайте исключения без нужды (возвращайте False). Для композиции используйте contextlib.ExitStack/AsyncExitStack, для потоков/соединений — async with/aclose. В корутинах без менеджера — try/finally. Критические закрытия при отмене можно кратко «экранировать» (shield) ради целостности.
Библиотека собеса по Python
Библиотека собеса по Python
Как делать атомарную запись и надёжную фиксацию файла при сбоях и конкурирующих записях?
Пишите во временный файл в том же каталоге, flush() → os.fsync(tmp) → os.replace(tmp, target) (атомарный rename), затем os.fsync(dir). Для множественных писателей используйте файловый lock (fcntl/msvcrt/portalocker) или координацию через очередь/БД; не перезаписывайте «на месте».
Библиотека собеса по Python
Библиотека собеса по Python