Как правильно управлять контекстом и очисткой в ​​асинхронном коде Python с помощью contextlib и asyncio? ⇐ Python

[Anonymous]

Я работал с асинхронным программированием на Python с использованием asyncio и столкнулся с проблемой при попытке управлять ресурсами, которые необходимо правильно очистить, даже несмотря на исключения или отмены.
Я понимаю, что contextlib предоставляет инструменты для управления ресурсами с помощью синхронных менеджеров контекста, но я не знаю, как правильно распространить это на асинхронные менеджеры контекста и обеспечить постоянную очистку ресурсов.
Вот пример того, чего я пытаюсь достичь:

Код: Выделить всё

import asyncio
import contextlib

class AsyncResource:
async def __aenter__(self):
print("Acquiring resource")
# Simulate acquiring resource
await asyncio.sleep(1)
return self

async def __aexit__(self, exc_type, exc, tb):
print("Releasing resource")
# Simulate releasing resource
await asyncio.sleep(1)

@contextlib.asynccontextmanager
async def async_resource_manager():
resource = AsyncResource()
await resource.__aenter__()
try:
yield resource
finally:
await resource.__aexit__(None, None, None)

async def main():
async with async_resource_manager() as resource:
print("Using resource")
# Simulate some work
await asyncio.sleep(2)

asyncio.run(main())

Это работает, как и ожидалось, но мне интересно:
Это идиоматический способ работы с асинхронными менеджерами контекста в Python?
Как я могу гарантировать правильную очистку ресурсов в случае исключения или отмены задачи?
Существуют ли какие-либо рекомендации или распространенные ошибки, о которых следует знать при использовании contextlib.asynccontextmanager с asyncio?
Любая информация или примеры были бы очень признательны!

Подробнее здесь: https://stackoverflow.com/questions/790 ... -with-cont