Cara mengatur atribut kelas dengan menunggu di __init__

Bagaimana saya bisa mendefinisikan kelas dengan awaitdalam konstruktor atau badan kelas?

Misalnya yang saya inginkan:

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'

atau contoh dengan atribut badan kelas:

class Foo(object):

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

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

foo = Foo(settings)

Solusi saya (Tapi saya ingin melihat cara yang lebih elegan)

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

Kebanyakan metode sihir tidak dirancang untuk bekerja dengan async def/ await- pada umumnya, Anda hanya harus menggunakan awaitdalam metode sihir asynchronous berdedikasi - __aiter__, __anext__, __aenter__, dan __aexit__. Menggunakannya di dalam metode ajaib lain tidak akan berfungsi sama sekali, seperti halnya __init__(kecuali Anda menggunakan beberapa trik yang dijelaskan dalam jawaban lain di sini), atau akan memaksa Anda untuk selalu menggunakan apa pun yang memicu panggilan metode ajaib dalam konteks asinkron.

asyncioPerpustakaan yang ada cenderung menangani ini dengan salah satu dari dua cara: Pertama, saya telah melihat pola pabrik yang digunakan ( asyncio-redis, misalnya):

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)

Pustaka lain menggunakan fungsi coroutine tingkat atas yang membuat objek, bukan metode pabrik:

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_poolFungsi dari aiopgyang Anda ingin menyebutnya dalam __init__benar-benar menggunakan pola yang tepat ini.

Ini setidaknya mengatasi __init__masalah tersebut. Saya belum melihat variabel kelas yang membuat panggilan asinkron di alam liar yang dapat saya ingat, jadi saya tidak tahu bahwa pola mapan apa pun telah muncul.

Cara lain untuk melakukan ini, untuk kesenangan:

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

Saya akan merekomendasikan metode pabrik terpisah. Aman dan mudah. Namun, jika Anda bersikeras menggunakan asyncversi __init__(), berikut ini contohnya:

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

Pemakaian:

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

Keluaran:

<class '__main__.Foo'>
True

Penjelasan:

Konstruksi kelas Anda harus menghasilkan coroutineobjek, bukan instance-nya sendiri.

Lebih baik lagi Anda bisa melakukan sesuatu seperti ini, yang sangat mudah:

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

Pada dasarnya yang terjadi di sini __init__() disebut pertama seperti biasa. Kemudian __await__()dipanggil yang kemudian menunggu async_init().

[Hampir] jawaban kanonik oleh @ojii

@dataclass
class Foo:
    settings: Settings
    pool: Pool

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