How to set class attribute with await in __init__

Question

How can I define a class with await in the constructor or class body?

For example what I want:

import asyncio

# some code


class Foo(object):

    async def __init__(self, settings):
        self.settings = settings
        self.pool = await create_pool(dsn)

foo = Foo(settings)
# it raises:
# TypeError: __init__() should return None, not 'coroutine'

or example with class body attribute:

class Foo(object):

    self.pool = await create_pool(dsn)  # Sure it raises syntax Error

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

foo = Foo(settings)

My solution (But I would like to see a more elegant way)

class Foo(object):

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

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

foo = Foo(settings)
await foo.init()

Answer 1

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.

Answer 2

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

Answer 3

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

Answer 4

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.

Answer 5

The AsyncObj class with __ainit__ "async-constructor":

class AsyncObj:
    def __init__(self, *args, **kwargs):
        """
        Standard constructor used for arguments pass
        Do not override. Use __ainit__ instead
        """
        self.__storedargs = args, kwargs
        self.async_initialized = False

    async def __ainit__(self, *args, **kwargs):
        """ Async constructor, you should implement this """

    async def __initobj(self):
        """ Crutch used for __await__ after spawning """
        assert not self.async_initialized
        self.async_initialized = True
        await self.__ainit__(*self.__storedargs[0], **self.__storedargs[1])  # pass the parameters to __ainit__ that passed to __init__
        return self

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

    def __init_subclass__(cls, **kwargs):
        assert asyncio.iscoroutinefunction(cls.__ainit__)  # __ainit__ must be async

    @property
    def async_state(self):
        if not self.async_initialized:
            return "[initialization pending]"
        return "[initialization done and successful]"

Here is example of "async class":

class MyAsyncObject(AsyncObj):
    async def __ainit__(self, param1, param2=0):
        print("hello!", param1, param2)
        # go something async, e.g. go to db
    

Usage:

async def example():
    my_obj = await MyAsyncObject("test", 123)

Answer 6

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

Answer 7

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

Answer 8

Vishnu shettigar's answer is so far the simplest, except that his async_init method doesn't return the object itself so foo isn't assigned a Foo instance. As for OP's purpose, the most elegant way to construct the class IMHO is

import asyncio

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

    def __await__(self):
        self.pool = asyncio.create_task(create_pool(dsn))
        yield from self.pool
        self.pool = self.pool.result()
        return self

To initialize the object, do the following

def main():
    loop = asyncio.get_event_loop()
    foo = loop.run_until_complete(Foo(settings))

Or

async def main():
    foo = await Foo(settings)

Answer 9

we could convert the async call to sync call by running the async code manually through asyncio.run()

class Foo:
    async def __ainit__(self, param):
        self._member = await some_async_func(param)

    def __init__(self, param):
        asyncio.run(self.__ainit__(param))

Answer 10

Just for records, this could also be achieved by creating a custom metaclass and overriding __call__ method. When you put parenthesis in front of the name of a class, you're actually calling this method. This is much like the solution suggested by @khazhyk, except you interrupt the procedure earlier and also no need to override __new__.

# interpreter version: 3.10
import asyncio


class AsyncInit(type):
    async def __call__(cls, *args, **kwargs):
        instance = cls.__new__(cls, *args, **kwargs)
        await instance.__init__(*args, **kwargs)
        return instance


class A(metaclass=AsyncInit):
    async def __init__(self, n):
        print(f"{n=} - giving the control back to event loop")
        await asyncio.sleep(2)
        print("initialization finished")


async def main():
    await asyncio.gather(*[A(i) for i in range(1, 5)])


asyncio.run(main())

output:

n=1 - giving the control back to event loop
n=2 - giving the control back to event loop
n=3 - giving the control back to event loop
n=4 - giving the control back to event loop
initialization finished
initialization finished
initialization finished
initialization finished

Answer 11

Depending on your needs, you can also use AwaitLoader from: https://pypi.org/project/async-property/

From the docs:

AwaitLoader will call await instance.load(), if it exists, before loading properties.

Answer 12

Everyone can try： https://pypi.org/project/asyncinit/

• pip install asyncinit

from asyncinit import asyncinit

@asyncinit
class MyClass:
    async def __init__(self, param):
        self.val = await self.deferredFn(param)

    async def deferredFn(self, x):
        # ...
        return x + 2

obj = await MyClass(42)
assert obj.val == 44

Answer 13

I wrote this mixin:

import asyncio


class AsyncMixin:
    """Adds an async_init method to an object which is called when the
    object is awaited.
    Typically the idiom obj = await Object()
    provides the synchronous __init__() and async async_init() calls"""

    async def async_init(self):
        """If an AsyncMixin object is created in an async context (ie await
        Object() then the __init__ method is non-async as normal but
        the async_init() method is called immediately by the
        __await__() magic method.
        """
        pass

    async def _async_init(self):
        task = asyncio.create_task(self.async_init())
        await task
        return self

    def __await__(self):
        return self._async_init().__await__()
    pass

So the OP's solution becomes:

class Foo(object, AsyncMixin):

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

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

foo = await Foo(settings)

Which I think is quite elegant

Answer 14

This worked for me in Python 3.9

from aiobotocore.session import AioSession
import asyncio




class SomeClass():

    def __init__(self):
        asyncio.run(self.async_init())
        print(self.s3)

    async def async_init(self):
        self.s3 = await AioSession().create_client('s3').__aenter__()