Блог

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

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

 class IOResourceInterface:
    def __init__(self, url, config={}):
        # Normal class initialization
        # store URL(s) to resources (such as files, or Network bound resources)
        # Store Configurations (like database creds, etc.)
        pass

    async def init_coro(self):
        # Eagerly Initializing Connection to External Resources
        await asyncio.sleep(1) # simulates the initialization of IO Resources
        
    
    def __await__(self) -> "IOResourceInterface":
        yield from self.init_coro().__await__()
        return self

    async def do_stuff(self):
        pass

# And then within the event loop
resource = await IOResourceInterface("protocol://resource",config={"user":"", "pass":""})

# Here resource is fully initialized and ready to go
await resource.do_stuff()
 

Какой подход использовать, всегда зависит от назначения класса и окружающего кода.

Но я предпочитаю два подхода:

• Заводским способом. В этом случае метод асинхронного класса выполняет всю необходимую инициализацию и передает инициализированные объекты __init__ методу в виде внедрения зависимостей:

 class IOResourceInterface:
    def __init__(self, initialized_resources):
        pass

    async def do_stuff(self):
        pass

    @classmethod
    async def create(cls, url, config):
        # Eagerly Initializing Connection to External Resources
        await asyncio.sleep(1)  # simulates the initialization of IO Resources
        initialized_resources = {}
        return cls(initialized_resources)


io = await IOResourceInterface.create("protocol://resource", config={})
await io.do_stuff()
 

• Асинхронный контекстный менеджер. Если класс требует не только инициализации, но и прямой деинициализации (закрытие соединений, очистка ресурсов и т. Д.), Часто бывает полезно сделать его асинхронным контекстным менеджером с использованием методов __aenter__ и. __aexit__ Вся работа с экземпляром класса выполняется в блоке контекстного менеджера:

 class IOResourceInterface:
    def __init__(self, url, config):
        pass

    async def __aenter__(self):
        await asyncio.sleep(1)  # simulates the initialization of IO Resources
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await asyncio.sleep(1)  # simulates cleaning resources

    async def do_stuff(self):
        pass

async with IOResourceInterface("protocol://resource", config={}) as io:
    await io.do_stuff()