How to set class attribute with await in __init__

Question

Most magic methods aren't designed to work with async def/await - in general, you should only be using await inside the dedicated asynchronous magic methods - __aiter__, __anext__, __aenter__, and __aexit__. Using it inside other magic methods either won't work at all, as is the case with __init__ (unless you use some tricks described in other answers here), or will force you to always use whatever triggers the magic method call in an asynchronous context.

Existing asyncio libraries tend to deal with this in one of two ways: First, I've seen the factory pattern used (asyncio-redis, for example):

import asyncio

dsn = "..."

class Foo(object):
    @classmethod
    async def create(cls, settings):
        self = Foo()
        self.settings = settings
        self.pool = await create_pool(dsn)
        return self

async def main(settings):
    settings = "..."
    foo = await Foo.create(settings)

Other libraries use a top-level coroutine function that creates the object, rather than a factory method:

import asyncio

dsn = "..."

async def create_foo(settings):
    foo = Foo(settings)
    await foo._init()
    return foo

class Foo(object):
    def __init__(self, settings):
        self.settings = settings

    async def _init(self):
        self.pool = await create_pool(dsn)

async def main():
    settings = "..."
    foo = await create_foo(settings)

The create_pool function from aiopg that you want to call in __init__ is actually using this exact pattern.

This at least addresses the __init__ issue. I haven't seen class variables that make asynchronous calls in the wild that I can recall, so I don't know that any well-established patterns have emerged.

Another way to do this, for funsies:

class aobject(object):
    """Inheriting this class allows you to define an async __init__.

    So you can create objects by doing something like `await MyClass(params)`
    """
    async def __new__(cls, *a, **kw):
        instance = super().__new__(cls)
        await instance.__init__(*a, **kw)
        return instance

    async def __init__(self):
        pass

#With non async super classes

class A:
    def __init__(self):
        self.a = 1

class B(A):
    def __init__(self):
        self.b = 2
        super().__init__()

class C(B, aobject):
    async def __init__(self):
        super().__init__()
        self.c=3

#With async super classes

class D(aobject):
    async def __init__(self, a):
        self.a = a

class E(D):
    async def __init__(self):
        self.b = 2
        await super().__init__(1)

# Overriding __new__

class F(aobject):
    async def __new__(cls):
        print(cls)
        return await super().__new__(cls)

    async def __init__(self):
        await asyncio.sleep(1)
        self.f = 6

async def main():
    e = await E()
    print(e.b) # 2
    print(e.a) # 1

    c = await C()
    print(c.a) # 1
    print(c.b) # 2
    print(c.c) # 3

    f = await F() # Prints F class
    print(f.f) # 6

import asyncio
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

I would recommend a separate factory method. It's safe and straightforward. However, if you insist on a async version of __init__(), here's an example:

def asyncinit(cls):
    __new__ = cls.__new__

    async def init(obj, *arg, **kwarg):
        await obj.__init__(*arg, **kwarg)
        return obj

    def new(cls, *arg, **kwarg):
        obj = __new__(cls, *arg, **kwarg)
        coro = init(obj, *arg, **kwarg)
        #coro.__init__ = lambda *_1, **_2: None
        return coro

    cls.__new__ = new
    return cls

Usage:

@asyncinit
class Foo(object):
    def __new__(cls):
        '''Do nothing. Just for test purpose.'''
        print(cls)
        return super().__new__(cls)

    async def __init__(self):
        self.initialized = True

async def f():
    print((await Foo()).initialized)

loop = asyncio.get_event_loop()
loop.run_until_complete(f())

Output:

<class '__main__.Foo'>
True

Explanation:

Your class construction must return a coroutine object instead of its own instance.

Better yet you can do something like this, which is very easy:

import asyncio

class Foo:
    def __init__(self, settings):
        self.settings = settings

    async def async_init(self):
        await create_pool(dsn)

    def __await__(self):
        return self.async_init().__await__()

loop = asyncio.get_event_loop()
foo = loop.run_until_complete(Foo(settings))

Basically what happens here is __init__() gets called first as usual. Then __await__() gets called which then awaits async_init().

[Almost] canonical answer by @ojii

@dataclass
class Foo:
    settings: Settings
    pool: Pool

    @classmethod
    async def create(cls, settings: Settings, dsn):
        return cls(settings, await create_pool(dsn))

I would like to show a much easier way of initiating coroutine based method within the __init__ method.

import asyncio


class Foo(object):

    def __init__(self, settings):
        self.settings = settings
        loop = asyncio.get_event_loop() 
        self.pool = loop.run_until_complete(create_pool(dsn))

foo = Foo(settings)

Important point to be noted is:

• This makes the async code work as sync(blocking)
• This is not the best way to run async code, but when it comes to only initiation via a sync method eg: __init__ it will be a good fit.
• After initiation, you can run the async methods from the object with await. i.e await foo.pool.get(value)
• Do not try to initiate via an await call you will get RuntimeError: This event loop is already running