3

Digamos que eu tenha uma classe em que preciso inicializar um campo com o retorno de uma corotina:

import asyncio

class Server:

    def __init__(self):
        self.connection = await self.connect()

    async def connect(self):
        await asyncio.sleep(1)
        return 'connection'

Fazendo isso é esperado que tenho um erro, visto que estou utilizado await dentro de um método que não é async. Também não é possível apenas adicionar o async:

class Server:

    async def __init__(self):
        self.connection = await self.connect()

    async def connect(self):
        await asyncio.sleep(1)
        return 'connection'

Pois assim __init__ passaria a retornar uma corotina e não None, como deveria. Uma alternativa seria definir um método de classe para isso:

class Server:

    async def connect(self):
        await asyncio.sleep(1)
        return 'connection'

    @classmethod
    async def initialize(cls):
        obj = cls()
        obj.connection = await obj.connect()
        return obj

Podendo chamar server = await Server.initialize(), mas com __init__ não é possível?

2
  • Interessante a pergunta e a resposta excelente do @jsbueno. Mas a pergunta que não quer calar: porque criar uma classe nesse caso? Quase sempre um Server é um singleton, e #Python já vem com o padrão singleton pronto para usar: chama-se módulo. Em vez de métodos, funções. Em vez de atributos, variáveis no escopo do módulo. Um módulo é mais simples que uma classe, e resolve tão bem quanto. E nunca será carregado mais de uma vez em memória. O runtime garante isso. 11/09/2020 às 20:02
  • @LucianoRamalho Para ser sincero, não lembro na época qual era a intenção, mas não era nada prático. Se não me engano eu estava estudando uma aplicação de corotinas em um serviço Web e me surgiu essa dúvida, porque imaginava que daria para fazer algo a partir do método __new__.
    – Woss
    11/09/2020 às 21:35

1 Resposta 1

3

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.

2
  • Acho que esse exemplo ficou bom o suficiente para prod. se essa necessidade for real - coloquei num gist por enquanto - gist.github.com/jsbueno/32beb7e087176089b8c33b479cb22129
    – jsbueno
    13/06/2019 às 3:12
  • A dúvida surgiu meio que aleatoriamente de um teste e não consegui definir sozinho que faria sentido ou não e se existiria uma justificativa para não existir um inicializador async, por isso a pergunta. Não achei justificativa para não permitir fazer isso e pela resposta parece ser algo plausível.
    – Woss
    13/06/2019 às 12:45

Você deve fazer log-in para responder a esta pergunta.

Esta não é a resposta que você está procurando? Pesquise outras perguntas com a tag .