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

The goal is to define and initialize a python class that acts as an interface to resources that involve blocking IO Operations during initialization, and normal program operation.

Based on a few other posts here, The best that I could come up with is as below, Is there a better way, and if not, why not?

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

Answer

Which approach to use always depends on purpose of the class and the surrounding code.

But I prefer two approaches:

  • Factory method. In the case, asynchronous class method carry out all the necessary initialization and pass the initialized objects to the __init__ method as dependency injection:
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()
  • Asynchronous context manager. If a class requires not only initialization, but also direct deinitialization (closing connections, cleaning up resources and etc.), it is often useful to make it as an asynchronous context manager using methods __aenter__ and __aexit__. All work with the class instance is in context manager block:
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()