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

Subcontinent

You might have some luck with __new__, although it might not be elegant – Bullfrog

I don't have experience with 3.5, and in other languages this wouldn't work because of the viral nature of async/await, but have you tried defining an async function like _pool_init(dsn) and then calling it from __init__? It would preserve the init-in-constructor appearance. – Benzoin

If you use curio: curio.readthedocs.io/en/latest/… – Silvertongued

use @classmethod 😎 it's an alternate constructor. put the async work there; then in __init__, just set the self attributes – Rager

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

Markman

Thanks @dano, I like the asyncio-redis approach and I am trying to copy it for my code. My problem is that I'd need to inherit Foo in another class. I am used to inheriting classes with super().__init__(*parent_args, **parent_kwargs), but I can't figure out how to both init the parent and call the factory method. Doing super().__init__() and then Parent.create() does not work – Ashurbanipal

@dano, you should fix up your first example code block here. Foo.create is a classmethod but it is not using cls. That self = Foo() should be self = cls(). – Bose

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

Annieannihilate

This is currently the most clear and understandable implementation in my opinion. I really like how intuitively extensible it is. I was worried it would be necessary to delve into metaclasses. – Fogel

This doesn't have correct __init__ semantics if super().__new__(cls) returns a pre-existing instance - normally, this would skip __init__, but your code not. – Livery

Hmm, per object.__new__ documentation, __init__ should only be invoked if isinstance(instance, cls) ? This seems somewhat unclear to me... But I don't see the semantics you claim anywhere... – Annieannihilate

Thinking about this more, if you override __new__ to return a pre-existing object, that new would need to be the outermost to make any sense, since other implementations of __new__ would have no general way of knowing if you're returning a new uninitialized instance or not. – Annieannihilate

Can @tankobot or Khazhyk or someone help explain this magic? Why does that async __new__(...) magically makes async __init__(...) valid? – Cryptonym

@Cryptonym because the __new__ method awaits __init__, and there's nothing preventing you from defining any of these magic methods as async – Annieannihilate

@Annieannihilate Well, there IS definitely something preventing you from defining async def __init__(...), as showed by the OP, and I believe that TypeError: __init__() should return None, not 'coroutine' exception is hardcoded inside Python and could not be bypassed. So I tried to understand how an async def __new__(...) made a difference. Now my understanding is that, your async def __new__(...) (ab)use the characteristic of "if __new__() does not return an instance of cls, then __init__() will not be invoked". Your new __new__() returns a coroutine, not a cls. That's why. Clever hack! – Cryptonym

@Cryptonym ha, interesting! I stumbled past that then, learned something new today :) – Annieannihilate

@Annieannihilate technically you don't need to make any magic methods async, you just need __new__ to return an Awaitable instead of an instance. So you can make an async class method helper and use that inside __new__ if you wanted. You can always just call an async method without await, it returns a coroutine. Then you return the coroutine from __new__ and let someone else await it to get the instance. – Syllabize

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

Alvord

I like this simple and straight design for initialize a async class but I think you forget to write return self in async_init() method – Anis

This is probably the best solution, if you want to use base classes or class decorators that rely on standard __new__() and __init__(). E.g. the popular dataclasses. The only issue I can see is that the instance can be awaited more than once, calling the __await__() more than once. But somebody may intentionally want this. Or if it is an issue, it can be prevented by a flag. – Laresa

async def __aenter__(self): return await self is missing, to make this usable as an async context manager like async with Foo() as foo: foo.bar() – Genetic

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.

Coloration

Couldn't you name your new __new__ and use super (likewise for __init__, i.e. just let the client override that) instead? – Calefacient

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)

Pushbike

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

Lehmann

dataclasses for the win! so easy. – Rager

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

Huck

in other words, this solution works except anywhere you'd actually want to use it, i.e. when the loop is already running. And what about anyio / trio? – Calefacient

@MatthiasUrlichs I dont know much about trio. In case of running loop, you can push this co-routine as a task and wait for it to be completed. – Huck

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)

Unmanned

This solution unfortunately allows a spurious "await object" to actually run, which is not what you'd intend to do if you're writing resilient programs. Alternately if I do want to add an __await__ method to the class, this solution prevents me from doing that. – Calefacient

@Matthias, could you tell another use of await magic method in the class? I can only see c = await C(), and this is it. – Laresa

The problem arises if you then mistakenly write "await c" instead of "await c.wait()" or similar. No checker will report that error. At minimum I'd throw an error if self.pool already exists. – Calefacient

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

Penton

No you can't. (a) nobody said you're using asyncio, (b) calling asyncio.run` recursively doesn't work. – Calefacient

Answer 10

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

Marozas

Answer 11

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

Knickknack

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

Carlsbad

As it’s currently written, your answer is unclear. Please edit to add additional details that will help others understand how this addresses the question asked. You can find more information on how to write good answers in the help center. – Tass

Answer 13

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.

Asmodeus

Answer 14

another trick to override an existing class with sync init and async methods

#!/usr/bin/env python3

# lazy async init
# this only works if all methods are async

import asyncio

class LazyAsyncReader:
    def __init__(self):
        self.value = "ainit ..."
        self._ainit_done = False
    async def __ainit__(self):
        self.value = "ainit done"
        self._ainit_done = True
    async def read(self):
        if not self._ainit_done:
            await self.__ainit__()
        await asyncio.sleep(0)
        return self.value

# alternative: unswitch the read function by monkeypatching
class LazyAsyncReader:
    def __init__(self):
        self.value = "ainit ..."
    async def __ainit__(self):
        self.value = "ainit done"
    async def read(self, *a, **k):
        await self.__ainit__()
        self.read = self._read_ainit_done
        return await self._read_ainit_done(*a, **k)
    async def _read_ainit_done(self):
        await asyncio.sleep(0)
        return self.value

async def main():
    # sync init
    reader = LazyAsyncReader()
    # async method
    res = await reader.read()
    print(res)

asyncio.run(main())

Genetic

Answer 15

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

Nesline

No it doesn't. You can't just nest asyncio.run calls. Also what if I'm using anyio ?? – Calefacient

I just did. what's up! – Nesline