correct/best 定义和初始化 python class who init 涉及 Asyncio 例程的方法是什么

What is the correct/best way to define and initialize a python class who init involves Asyncio routines

目标是定义和初始化一个 python class 作为资源的接口,在初始化和正常程序操作期间涉及阻塞 IO 操作。

根据这里的其他一些帖子,我能想到的最好的方法如下,有没有更好的方法,如果没有,为什么不呢?

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()

使用哪种方法始终取决于 class 和周围代码的目的。

但我更喜欢两种方法:

  • 工厂方法。在这种情况下,异步 class 方法执行所有必要的初始化并将初始化的对象作为依赖注入传递给 __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()
  • 异步上下文管理器。如果 class 不仅需要初始化,还需要直接取消初始化(关闭连接、清理资源等),使用方法 __aenter__ 将其设为 asynchronous context manager__aexit__。 class 实例的所有工作都在上下文管理器块中:
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()