如何在 __init__ 中使用 await 设置类属性答案

【问题标题】：How to set class attribute with await in __init__如何在 __init__ 中使用 await 设置类属性
【发布时间】：2016-01-12 17:14:25
【问题描述】：

如何在构造函数或类主体中定义带有await 的类？

例如我想要的：

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'

或带有类主体属性的示例：

class Foo(object):

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

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

foo = Foo(settings)

我的解决方案（但我希望看到更优雅的方式）

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

【问题讨论】：

__new__ 可能会带来一些运气，尽管它可能并不优雅
我没有使用 3.5 的经验，而在其他语言中，由于 async/await 的病毒性质，这不起作用，但是您是否尝试过定义像 _pool_init(dsn) 这样的异步函数，然后从__init__ 调用它？它将保留 init-in-constructor 的外观。
如果你使用古玩：curio.readthedocs.io/en/latest/…
使用@classmethod ????它是一个替代构造函数。把异步工作放在那里；然后在__init__ 中，只需设置self 属性

标签： python python-3.x python-asyncio

【解决方案1】：

大多数魔术方法并非设计用于async def/await - 通常，您应该只在专用异步魔术方法中使用await - __aiter__、__anext__、__aenter__和__aexit__。在其他魔术方法中使用它要么根本不起作用，就像__init__ 的情况一样（除非您使用此处其他答案中描述的一些技巧），或者会迫使您始终使用触发魔术方法调用的任何东西异步上下文。

现有的asyncio 库倾向于通过以下两种方式之一来处理这个问题：首先，我看到了使用的工厂模式（例如asyncio-redis）：

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)

其他库使用创建对象的顶级协程函数，而不是工厂方法：

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)

您要在__init__ 中调用的来自aiopg 的create_pool 函数实际上正在使用这种精确模式。

这至少解决了__init__ 的问题。我还没有看到我记得的在野外进行异步调用的类变量，所以我不知道是否出现了任何成熟的模式。

【讨论】：

【解决方案2】：

另一种方法，有趣的是：

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

【讨论】：

这是目前我认为最清晰易懂的实现。我真的很喜欢它的直观可扩展性。我担心有必要深入研究元类。
如果 super().__new__(cls) 返回一个预先存在的实例，则这不具有正确的 __init__ 语义 - 通常，这会跳过 __init__，但您的代码不会。
嗯，根据 object.__new__ 文档，只有在 isinstance(instance, cls) 时才应该调用 __init__？这对我来说似乎有点不清楚......但我没有看到你在任何地方声称的语义......
再想一想，如果你重写 __new__ 以返回一个预先存在的对象，那么 new 必须是最外层才有意义，因为 __new__ 的其他实现将没有了解您是否返回新的未初始化实例的一般方法。
@khazhyk 好吧，正如 OP 所示，肯定有一些东西阻止你定义 async def __init__(...)，我相信 TypeError: __init__() should return None, not 'coroutine' 异常在 Python 中是硬编码的，无法绕过。所以我试图了解async def __new__(...) 是如何产生影响的。现在我的理解是，您的async def __new__(...) (ab) 使用“如果__new__() 不返回cls 的实例，则__init__() 将不会被调用”的特性。您的新 __new__() 返回协程，而不是 cls。这就是为什么。聪明的黑客！

【解决方案3】：

我会推荐一个单独的工厂方法。这是安全和直接的。但是，如果您坚持使用__init__() 的async 版本，这里有一个示例：

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

用法：

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

输出：

<class '__main__.Foo'>
True

说明：

你的类构造必须返回一个coroutine 对象而不是它自己的实例。

【讨论】：

您不能命名您的new __new__ 并使用super（同样适用于__init__，即让客户覆盖它）吗？

【解决方案4】：

更好的是你可以做这样的事情，这很容易：

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

基本上这里发生的是__init__() 像往常一样首先被调用。然后__await__() 被调用，然后等待async_init()。

【讨论】：

【解决方案5】：

[几乎]@ojii 的规范回答

@dataclass
class Foo:
    settings: Settings
    pool: Pool

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

【讨论】：

dataclasses 赢得胜利！这么简单。

【解决方案6】：

我想展示一种在 __init__ 方法中启动基于协程的方法的更简单方法。

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)

需要注意的重点是：

这使得异步代码作为同步（阻塞）工作
这不是运行异步代码的最佳方式，但如果仅通过同步方法启动，例如：__init__，它将非常适合。
启动后，您可以使用 await 从对象运行异步方法。即await foo.pool.get(value)
不要尝试通过await 发起呼叫，您将收到RuntimeError: This event loop is already running

【讨论】：

【解决方案7】：

带有__ainit__ "async-constructor" 的 AsyncObj 类：

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]"

这是“异步类”的示例：

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

用法：

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

【讨论】：

【解决方案8】：

到目前为止，Vishnu shettigar 的答案是最简单的，除了他的 async_init 方法不返回对象本身，因此 foo 没有分配 Foo 实例。至于OP的目的，构造类恕我直言的最优雅的方式是

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

要初始化对象，请执行以下操作

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

或者

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

【讨论】：

【解决方案9】：

根据您的需要，您还可以使用来自以下地址的AwaitLoader： https://pypi.org/project/async-property/

来自文档：

AwaitLoader 将在加载属性之前调用 await instance.load()（如果存在）。

【讨论】：

【解决方案10】：

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

【讨论】：

【解决方案11】：

我们可以通过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))

【讨论】：