Cómo establecer el atributo de clase con await en __init__

¿Cómo puedo definir una clase awaiten el constructor o en el cuerpo de la clase?

Por ejemplo lo que quiero:

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'

o ejemplo con el atributo del cuerpo de la clase:

class Foo(object):

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

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

foo = Foo(settings)

Mi solución (pero me gustaría ver una forma más elegante)

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 mayoría de los métodos mágicos no están diseñados para trabajar con async def/ await- en general, sólo se debe a utilizar awaitdentro de los métodos mágicos dedicados asíncronos - __aiter__, __anext__, __aenter__, y __aexit__. Usarlo dentro de otros métodos mágicos no funcionará en absoluto, como es el caso con __init__(a menos que use algunos de los trucos descritos en otras respuestas aquí), o lo obligará a usar siempre lo que desencadena la llamada al método mágico en un contexto asincrónico.

Las asynciobibliotecas existentes tienden a lidiar con esto de una de dos maneras: Primero, he visto el patrón de fábrica utilizado ( asyncio-redispor ejemplo):

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)

Otras bibliotecas utilizan una función de corrutina de nivel superior que crea el objeto, en lugar de un método de fábrica:

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_poolfunción desde la aiopgque desea llamar __init__está usando este patrón exacto.

Esto al menos aborda el __init__problema. No he visto variables de clase que hagan llamadas asincrónicas en la naturaleza que pueda recordar, por lo que no sé si han surgido patrones bien establecidos.

Otra forma de hacer esto, por diversión:

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

Recomendaría un método de fábrica por separado. Es seguro y sencillo. Sin embargo, si insiste en una asyncversión de __init__(), aquí tiene un ejemplo:

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

Uso:

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

Salida:

<class '__main__.Foo'>
True

Explicación:

La construcción de su clase debe devolver un coroutineobjeto en lugar de su propia instancia.

Mejor aún, puede hacer algo como esto, que es muy fácil:

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

Básicamente, lo que sucede aquí es que __init__() se llama primero como de costumbre. Luego __await__()se llama que luego espera async_init().

[Casi] respuesta canónica de @ojii

@dataclass
class Foo:
    settings: Settings
    pool: Pool

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