如何在 __init__ 中使用 await 设置 class 属性
How to set class attribute with await in __init__
如何在构造函数或 class 主体中使用 await
定义 class?
比如我想要的:
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 正文属性的示例:
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()
大多数魔术方法并非设计用于 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__
问题。我还没有看到 class 我记得的在野外进行异步调用的变量,所以我不知道是否出现了任何公认的模式。
我会推荐一个单独的工厂方法。它既安全又直接。但是,如果您坚持使用 __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
解释:
您的 class 构造必须 return 一个 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/
来自文档:
AwaitLoader
will call await instance.load()
, if it exists, before loading properties.
我想展示一种更简单的方法来在 __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
具有 __ainit__
“async-constructor”的 AsyncObj class:
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”的例子:
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
方法不 return 对象本身,所以 foo
没有分配 Foo
实例。至于 OP 的目的,构建 class 恕我直言的最优雅方法是
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/
- pip 安装 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
如何在构造函数或 class 主体中使用 await
定义 class?
比如我想要的:
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 正文属性的示例:
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()
大多数魔术方法并非设计用于 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__
问题。我还没有看到 class 我记得的在野外进行异步调用的变量,所以我不知道是否出现了任何公认的模式。
我会推荐一个单独的工厂方法。它既安全又直接。但是,如果您坚持使用 __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
解释:
您的 class 构造必须 return 一个 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/
来自文档:
AwaitLoader
will call awaitinstance.load()
, if it exists, before loading properties.
我想展示一种更简单的方法来在 __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
具有 __ainit__
“async-constructor”的 AsyncObj class:
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”的例子:
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
方法不 return 对象本身,所以 foo
没有分配 Foo
实例。至于 OP 的目的,构建 class 恕我直言的最优雅方法是
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/
- pip 安装 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