时间戳

首页 > 解决方案 > 定义和初始化涉及 Asyncio 例程的 python 类的正确/最佳方法是什么

python - 定义和初始化涉及 Asyncio 例程的 python 类的正确/最佳方法是什么

问题描述

目标是定义和初始化一个 python 类，它充当资源的接口，涉及初始化期间阻塞 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()

标签： pythonpython-asyncio

解决方案

使用哪种方法总是取决于类的目的和周围的代码。

但我更喜欢两种方法：

  • 工厂方法。在这种情况下，异步类方法执行所有必要的初始化并将初始化的对象__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()

推荐阅读