Como você percebeu, não, não é possível de forma ordinária.
Então temos que usar os recusos extra-ordinários a disposição na linguagem.
Transformar uma função em async ou num generator é bem menos inocente do que parece - a função muda fundamentalmente, fica só "parecendo" uma função.
O melhor workaround parece ser algo como:
- ter as chamadas async num outro método que não o
__init__
, como vocẽ colocou,
- Criar uma função assincrona externa, que funcione como "factory" para instâncias da sua classe, que chame o
__init__
de forma síncrona, e dê um await no método com inicialização assíncrona
- Usar um decorator ou uma metaclasse para colocar um mecanismo que permita com que o nome da classe sendo chamado como se fosse uma instância normal, retorne um objeto awaitable que faça a inicialização toda.
- o objeto awaitable retornado no passo anterior deve retornar a instância da classe, e não o valor retornado pelo "async init" original.
Na verdade, com as ideias acima, é possível sim montar uma metaclasse que aceite ter o __init__
como async, e possa trabalhar com isso - mas uma das coisas que a metaclasse vai fazer é justamente renomear o __init__
para outro nome.
É preciso uma metaclasse um pouco mais elaborada que a maioria delas, que costuma customizar apenas o método __new__
- e customizar o método __call__
. Embora pouco visto na parca documentaçao e exemplos de metaclasses, o __call__
da metaclasse é o método que orquestra a chamada ao __new__
e __init__
da classe, e retorna a instância (que foi inicialmente retornada pelo __new__
). O __call__
de type
, a metaclasse padrão, é em código nativo, mas a grosso modo, o que ele é equivalente a:
def __call__(cls, *args, **kw):
instance = cls.__new__(cls, *args, **kw)
if type(instance) is cls:
instance.__init__(*args, **kw)
return instance
Então deve ser possível fazer algo nesta linha:
import inspect
from functools import wraps
def async_init_wrapper(func):
def wrapper_stage1(instance):
async def wrapper_stage2(*args, **kw):
value = await func(instance, *args, **kw)
if value is not None:
raise TypeError("__async_init__() should return None")
return instance
return wrapper_stage2
wrapper_stage1.__name__ = func.__name__
return wrapper_stage1
class AwaitableClass(type):
def __new__(mcls, name, bases, ns, **kw):
if "__init__" in ns and inspect.iscoroutinefunction(ns["__init__"]):
ns["__async_init__"] = async_init_wrapper(ns.pop("__init__"))
return super().__new__(mcls, name, bases, ns, **kw)
def __call__(cls, *args, **kw):
instance = super().__call__(*args, **kw)
if not isinstance(instance, cls) or not hasattr(cls, "__async_init__"):
return instance
return instance.__async_init__()(*args, **kw)
def test_awaitable_class():
import asyncio
class Server(metaclass=AwaitableClass):
async def __init__(self):
self.connection = await self.connect()
async def connect(self):
await asyncio.sleep(1)
return 'server initialized connection'
async def concurrent_task():
await asyncio.sleep(0.5)
print("doing stuff while server is initialized")
async def init_server():
print("starting server initialization")
server_instance = await Server()
print("server ready")
return server_instance
async def main():
results = await asyncio.gather(
init_server(),
concurrent_task()
)
return results[0]
loop = asyncio.get_event_loop()
server_instance = loop.run_until_complete(main())
print(server_instance.connection)
if __name__ == "__main__":
test_awaitable_class()
A saida dessa listagem no terminal:
[gwidion@village tmp01]$ python3 awaitable_class.py
starting server initialization
doing stuff while server is initialized
server ready
server initialized connection
Note que há vários corner cases que nem chequei. Para dar um exemplo: criar um async __init__
desta forma, por exemplo, impede que dentro dele se chame super().__init__
.
O design que criei, quase que por sorte, vai funcionar no entanto com um __init__
sincrono e um async __async_init__
definido explicitamente no corpo da classe, no entanto - e o __init__
nesse caso pode fazer uso de super()
normalmente.
__new__
.