Comment définir l'attribut de classe avec await dans __init__

Comment puis-je définir une classe avec awaitdans le constructeur ou le corps de la classe?

Par exemple ce que je veux:

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'

ou exemple avec l'attribut de corps de classe:

class Foo(object):

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

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

foo = Foo(settings)

Ma solution (mais j'aimerais voir une manière plus élégante)

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

La plupart des méthodes magiques ne sont pas conçus pour fonctionner avec async def/ await- en général, vous ne devriez utiliser à l' awaitintérieur des méthodes magiques asynchrones dédiés - __aiter__, __anext__, __aenter__et __aexit__. Son utilisation dans d'autres méthodes magiques ne fonctionnera pas du tout, comme c'est le cas avec __init__(à moins que vous n'utilisiez certaines astuces décrites dans d'autres réponses ici), ou vous obligera à toujours utiliser tout ce qui déclenche l'appel de la méthode magique dans un contexte asynchrone.

Les asynciobibliothèques existantes ont tendance à gérer cela de deux manières: Premièrement, j'ai vu le modèle d'usine utilisé ( asyncio-redispar exemple):

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)

D'autres bibliothèques utilisent une fonction coroutine de niveau supérieur qui crée l'objet, plutôt qu'une méthode d'usine:

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)

La create_poolfonction à partir de aiopglaquelle vous souhaitez appeler __init__utilise en fait ce modèle exact.

Cela résout au moins le __init__problème. Je n'ai pas vu de variables de classe qui font des appels asynchrones dans la nature dont je puisse me souvenir, donc je ne sais pas si des modèles bien établis ont émergé.

Une autre façon de faire cela, pour le plaisir:

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

Je recommanderais une méthode d'usine distincte. C'est sûr et simple. Cependant, si vous insistez sur une asyncversion de __init__(), voici un exemple:

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

Production:

<class '__main__.Foo'>
True

Explication:

Votre construction de classe doit renvoyer un coroutineobjet au lieu de sa propre instance.

Mieux encore, vous pouvez faire quelque chose comme ça, ce qui est très simple:

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

Fondamentalement, ce qui se passe ici est __init__() appelé en premier, comme d'habitude. Puis __await__()est appelé qui attend ensuite async_init().

[Presque] réponse canonique de @ojii

@dataclass
class Foo:
    settings: Settings
    pool: Pool

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