synchronicity

O Python 3 tem um apoio incrível para a programação assíncrona, mas sem dúvida tornou um pouco mais difícil desenvolver bibliotecas. Você está cansado de implementar métodos síncronos e assíncronos fazendo basicamente a mesma coisa? Esta pode ser uma solução simples para você.

Instalação

 pip install synchronicity

Antecedentes: Por que algo assim é necessário

Digamos que você tenha uma função assíncrona

 async def f ( x ):
    await asyncio . sleep ( 1.0 )
    return x ** 2

E digamos (por qualquer motivo) que você queira oferecer uma API síncrona aos usuários. Por exemplo, talvez você queira facilitar a execução de seu código em um script básico, ou um usuário está criando algo que é principalmente ligado à CPU, para que eles não queiram se preocupar com o Asycio.

Uma maneira "simples" de criar um equivalente síncrono seria implementar um conjunto de funções síncronas onde tudo o que eles fazem é chamar asyncio.run em uma função assíncrona. Mas esta não é uma ótima solução para um código mais complexo:

• É um tipo de trabalho grunhido tedioso ter que fazer isso para cada método/função
• asyncio.run não funciona com geradores
• Em muitos casos, você precisa preservar um loop de eventos em execução entre as chamadas.

O último caso é particularmente desafiador. Por exemplo, digamos que você esteja implementando um cliente para um banco de dados que precisa ter uma conexão persistente e deseja construí -lo no Asycio:

 class DBConnection :
    def __init__ ( self , url ):
        self . _url = url

    async def connect ( self ):
        self . _connection = await connect_to_database ( self . _url )

    async def query ( self , q ):
        return await self . _connection . run_query ( q )

Como você expõe uma interface síncrona a este código? O problema é que o embrulho connect e query no Asyncio.run não funcionará, pois você precisa preservar o loop de eventos nas chamadas . É claro que precisamos de algo um pouco mais avançado.

Como usar esta biblioteca

Esta biblioteca oferece uma classe Synchronizer simples que cria um loop de eventos em um encadeamento separado e envolve funções/geradores/classes para que a execução síncrona aconteça nesse encadeamento. Quando você chama qualquer coisa, ele detectará se você estiver executando em um contexto síncrono ou assíncrono e se comportará correspondentemente.

• No caso síncrono, ele simplesmente bloqueará até que o resultado esteja disponível (observe que você pode fazê -lo retornar também um futuro, veja mais tarde)
• No caso assíncrono, ele funciona como o negócio usual de chamar o código assíncrono

 from synchronicity import Synchronizer

synchronizer = Synchronizer ()

@ synchronizer . create_blocking
async def f ( x ):
    await asyncio . sleep ( 1.0 )
    return x ** 2


# Running f in a synchronous context blocks until the result is available
ret = f ( 42 )  # Blocks
print ( 'f(42) =' , ret )


async def g ():
    # Running f in an asynchronous context works the normal way
    ret = await f ( 42 )
    print ( 'f(42) =' , ret )

Exemplos mais avançados

Geradores

O decorador também funciona em geradores:

 @ synchronizer . create_blocking
async def f ( n ):
    for i in range ( n ):
        await asyncio . sleep ( 1.0 )
	yield i


# Note that the following runs in a synchronous context
# Each number will take 1s to print
for ret in f ( 10 ):
    print ( ret )

Sincronizando classes inteiras

Ele também opera em aulas envolvendo todos os métodos da classe:

 @ synchronizer . create_blocking
class DBConnection :
    def __init__ ( self , url ):
        self . _url = url

    async def connect ( self ):
        self . _connection = await connect_to_database ( self . _url )

    async def query ( self , q ):
        return await self . _connection . run_query ( q )


# Now we can call it synchronously, if we want to
db_conn = DBConnection ( 'tcp://localhost:1234' )
db_conn . connect ()
data = db_conn . query ( 'select * from foo' )

Retornando futuros

Você também pode fazer com que as funções retornem um objeto Future adicionando _future=True a qualquer chamada. Isso pode ser útil se você deseja despachar muitas chamadas de um contexto de bloqueio, mas deseja resolvê -las aproximadamente em paralelo:

 from synchronicity import Synchronizer

synchronizer = Synchronizer ()

@ synchronizer . create_blocking
async def f ( x ):
    await asyncio . sleep ( 1.0 )
    return x ** 2

futures = [ f ( i , _future = True ) for i in range ( 10 )]  # This returns immediately
rets = [ fut . result () for fut in futures ]  # This should take ~1s to run, resolving all futures in parallel
print ( 'first ten squares:' , rets )

Usando com outro código assíncrono

Esta biblioteca também pode ser útil em configurações puramente assíncronas, se você tiver vários loops de eventos ou se tiver alguma seção que seja ligada à CPU ou em algum código crítico que deseja executar em um thread separado para segurança. Todas as chamadas para funções/geradores sincronizados são seguros por threads por design. Isso o torna uma alternativa útil para loop.run_in_executor para coisas simples. Observe, no entanto, que cada sincronizador executa apenas um thread.

Gerentes de contexto

Você pode sincronizar as classes do Gerenciador de contexto como qualquer outra classe e os métodos especiais serão tratados corretamente.

Há também um decorador de funções @synchronizer.asynccontextmanager que se comporta exatamente como https://docs.python.org/3/library/contextlib.html#contextlib.asynccontextmanager, mas trabalha em contextos síncronos e síncronos e assonetos.

Botchas

• Funciona para as aulas que são gerentes de contexto, mas não para funções que retornam um gerente de contexto
• Ele cria uma nova classe (com o mesmo nome) ao envolver aulas, o que pode levar a problemas de digitação se você tiver algum uso não sincronizado da mesma classe
• Não faço ideia de como isso interage com as anotações de digitação
• Se uma classe for "sincronizada", envolve todos os métodos da classe, mas isso normalmente significa que você não pode alcançar atributos e executar o código assíncrono: você pode receber erros como "anexado a um loop diferente"
• Observe que todo o código sincronizado será executado em um tópico diferente e um loop de eventos diferente, portanto, chamar o código pode ter uma menor sobrecarga extra

Todos

• Apoie o caso oposto, ou seja, você tem uma função de bloqueio/gerador/classe/objeto, e deseja chamá-lo de forma assíncrona (isso é relativamente simples de fazer para funções simples usando loop.run_in_executor, mas o Python não possui suporte interno para geradores , e seria bom transformar uma aula inteira
• Mais documentação
• Possibilitar a anotação de métodos seletivamente para retornar futuros
• Talvez possibilite sincronizar objetos em tempo real, não apenas aulas

Esta biblioteca é uma borda de amputador de membros

Este é o código que eu quebrei de projetos pessoais e não foi testado em batalha. Há uma pequena suíte de teste que você pode executar usando pytest.

Processo de liberação

Deve automatizar isso ...

• Faça uma nova release-XYZ da Main
• Versão Bump em PyProject.toml para XYZ
• Cometer essa mudança e criar um PR
• Mesclar o PR uma vez verde
• Checkout main
• git tag -a vX.YZ -m "* release bullets"
• Git Push -Tags
• TWINE_USERNAME=__token__ TWINE_PASSWORD="$PYPI_TOKEN_SYNCHRONICITY" make publish