Is it possible to define async as the initializer method of a Python class?

Question

Let’s say I have a class where I need to initialize a field with the return of a corotina:

import asyncio

class Server:

    def __init__(self):
        self.connection = await self.connect()

    async def connect(self):
        await asyncio.sleep(1)
        return 'connection'

Doing so is expected to have an error, since I am used await within a method that is not async. It is also not possible to just add the async:

class Server:

    async def __init__(self):
        self.connection = await self.connect()

    async def connect(self):
        await asyncio.sleep(1)
        return 'connection'

So __init__ would return a corotina and not None, as it should. An alternative would be to define a class method for this:

class Server:

    async def connect(self):
        await asyncio.sleep(1)
        return 'connection'

    @classmethod
    async def initialize(cls):
        obj = cls()
        obj.connection = await obj.connect()
        return obj

Calling server = await Server.initialize(), but with __init__ is not possible?

Answer 1

As you have noticed, no, it is not possible ordinarily. So we have to use the extra-ordinary rejections to disposition in language.

Turning a function into an async or a Generator is much less innocent than it seems - the function changes fundamentally, is only "seeming" a function.

The best workaround seems to be something like:

• have async calls in a method other than __init__, as you put it,
• Create an external asynchronous function, which acts as "Factory" for instances in your class, which calls the __init__ synchronously, and give a await in the asynchronous initialization method
• Use a Decorator or metaclass to place a mechanism that allows the class name to be called as if it were a normal instance, return a awaitable object that initializes the whole.
• the awaitable object returned in the previous step must return the class instance, not the value returned by the original "async init".

In fact, with the above ideas, it is possible to set up a metaclass that accepts having the __init__ as async, and can work with it - but one of the things the metaclass will do is to just rename the __init__ for another name.

It takes a metaclass a little more elaborate than most of them, which usually just customize the method __new__ - and customize the method __call__ . Although little seen in the meager documentation and examples of metaclasses, the __call__ of metaclass is the method that orchestrates the call to __new__ and __init__ class, and returns the instance (which was initially returned by __new__). The __call__ of type, the standard metaclass, is in native code, but roughly, which is equivalent to:

def __call__(cls, *args, **kw):
   instance = cls.__new__(cls, *args, **kw)
   if type(instance) is cls:
      instance.__init__(*args, **kw)
   return instance

Then it must be possible to do something on this line:

import inspect
from functools import wraps

def async_init_wrapper(func):
    def wrapper_stage1(instance):
        async def wrapper_stage2(*args, **kw):
            value = await func(instance, *args, **kw)
            if value is not None:
                raise TypeError("__async_init__() should return None")
            return instance
        return wrapper_stage2
    wrapper_stage1.__name__ = func.__name__
    return wrapper_stage1


class AwaitableClass(type):
    def __new__(mcls, name, bases, ns, **kw):
        if "__init__" in ns and inspect.iscoroutinefunction(ns["__init__"]):
            ns["__async_init__"] = async_init_wrapper(ns.pop("__init__"))
        return super().__new__(mcls, name, bases, ns, **kw)

    def __call__(cls, *args, **kw):
        instance = super().__call__(*args, **kw)
        if not isinstance(instance, cls) or not hasattr(cls, "__async_init__"):
            return instance
        return instance.__async_init__()(*args, **kw)


def test_awaitable_class():
    import asyncio

    class Server(metaclass=AwaitableClass):

        async def __init__(self):
            self.connection = await self.connect()

        async def connect(self):
            await asyncio.sleep(1)
            return 'server initialized connection'

    async def concurrent_task():
        await asyncio.sleep(0.5)
        print("doing stuff while server is initialized")

    async def init_server():
        print("starting server initialization")
        server_instance = await Server()
        print("server ready")
        return server_instance

    async def main():
        results = await asyncio.gather(
            init_server(),
            concurrent_task()
        )
        return results[0]
    loop = asyncio.get_event_loop()
    server_instance = loop.run_until_complete(main())
    print(server_instance.connection)


if __name__ == "__main__":
    test_awaitable_class()

The exit of this listing in the terminal:

[gwidion@village tmp01]$ python3 awaitable_class.py 
starting server initialization
doing stuff while server is initialized
server ready
server initialized connection

Note that there are several corner cases I haven’t even checked. To give you an example: create a async __init__ in this way, for example, it prevents super().__init__. The design I created, almost luckily, will work however with a __init__ synchronous and a async __async_init__ explicitly defined in the class body, however - and __init__ in that case may make use of super() normally.