如何设置等待在 _ _ init _ _ 的类属性

如何在构造函数或类主体中定义具有 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 body 属性的示例:

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()
87492 次浏览
小开
最佳答案

大多数魔法方法都不是为了与 async def/await一起工作而设计的——一般来说，您应该只在专用的异步魔法方法(__aiter__、 __anext__、 __aenter__和 __aexit__)中使用 await。在其他魔法方法中使用它要么根本不起作用，就像在 __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__问题。我还没有见过在野外进行异步调用的类变量，所以我不知道是否已经出现了任何成熟的模式。

小开

我建议采用单独的工厂方法。安全又简单。然而，如果你坚持使用 async版本的 __init__()，这里有一个例子:

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对象，而不是它自己的实例。

小开

另一种方法，为了好玩:

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

更好的是，你可以这样做，这是非常容易的:

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

小开

[几乎]标准答案@ojii

@dataclass
class Foo:
settings: Settings
pool: Pool


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

根据您的需要，您也可以使用 AwaitLoader: Https://pypi.org/project/async-property/

来自文件:

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

小开

我想展示一种在 __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)

需要注意的重点是:

  • 这使得异步代码作为 sync (阻塞)工作
  • 这不是运行异步代码的最佳方式，但是当它只通过同步方法启动时，例如: __init__，这将是一个很好的选择。
  • 启动之后，可以使用 wait. ie await foo.pool.get(value)从对象运行异步方法
  • 不要试图通过一个 await呼叫启动，你会得到 RuntimeError: This event loop is already running
小开

具有 __ainit__“异步构造函数”的 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)
小开

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

Vishnu shettigar 的答案是目前为止最简单的，除了他的 async_init方法不返回对象本身，所以 foo没有被分配 Foo实例。至于 OP 的目的，构造类 IMHO 的最优雅的方法是

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

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

每个人都可以尝试: Https://pypi.org/project/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

提交答案