👩🏽‍🔧 🎣 🧖🏿 O que a assincronia deve ser em Python 💣 ⛹🏾 ➗

Nos últimos anos, a palavra async- chave de programação assíncrona e a semântica permearam muitas linguagens de programação populares: JavaScript , Rust , C # e muitas outras . Claro, Python também tem async/await, foi introduzido no Python 3.5.

Neste artigo, quero discutir os problemas do código assíncrono, especular sobre alternativas e propor uma nova abordagem para oferecer suporte a aplicativos síncronos e assíncronos ao mesmo tempo.

Cor da função

Quando funções assíncronas são incluídas em uma linguagem de programação, ela essencialmente se divide em duas. As funções vermelhas aparecem (ou assíncronas) e algumas funções permanecem azuis (síncronas).

O principal problema é que as funções azuis não podem chamar as vermelhas, mas as vermelhas podem causar as azuis. Em Python, por exemplo, isso é parcialmente verdadeiro: as funções assíncronas só podem chamar funções sem bloqueio síncronas. Mas é impossível determinar a partir da descrição se a função está bloqueando ou não. Python é uma linguagem de script.

Essa divisão leva à divisão da linguagem em dois subconjuntos: síncrono e assíncrono. O Python 3.5 foi lançado há mais de cinco anos, mas asyncainda não é tão bem suportado quanto os recursos síncronos do Python.

Você pode ler mais sobre as cores das funções neste excelente artigo .

Código duplicado

Diferentes cores de funções significam duplicação de código na prática.

Imagine que você está desenvolvendo uma ferramenta CLI para recuperar o tamanho de uma página da web e deseja manter formas síncronas e assíncronas de fazer isso. Por exemplo, isso é necessário se você estiver escrevendo uma biblioteca e não souber como seu código será usado. E não se trata apenas de bibliotecas PyPI, mas também de nossas próprias bibliotecas com lógica comum para vários serviços, escritas, por exemplo, em Django e aiohttp. Embora, é claro, os aplicativos independentes sejam em sua maioria escritos apenas de forma síncrona ou assíncrona.

Vamos começar com o pseudocódigo síncrono:

def fetch_resource_size(url: str) -> int:
    response = client_get(url)
    return len(response.content)

Parece legal. Agora vamos dar uma olhada no análogo assíncrono:

async def fetch_resource_size(url: str) -> int:
    response = await client_get(url)
    return len(response.content)

Em geral, é o mesmo código, mas com a adição das palavras asynce await. E eu não inventei - compare os exemplos de código no tutorial em httpx:

Existe exatamente a mesma imagem.

Abstração e composição

Acontece que você precisa reescrever todo o código síncrono e organizar aqui e ali asynce awaitpara que o programa se torna assíncrona.

Dois princípios podem ajudar a resolver esse problema. Primeiro, vamos reescrever o pseudocódigo imperativo em funcional. Isso permitirá que você veja a imagem mais claramente.

def fetch_resource_size(url: str) -> Abstraction[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

Você pergunta o que é esse método .map, o que ele faz. É assim que a composição de abstrações complexas e funções puras ocorre em um estilo funcional. Isso permite que você crie uma nova abstração com um novo estado a partir de um existente. Suponha que ele client_get(url)retorne inicialmente Abstraction[Response]e a chamada .map(lambda response: len(response.content))converta a resposta na instância necessária Abstraction[int].

Torna-se claro o que fazer a seguir. Observe como facilmente passamos de algumas etapas independentes para chamadas de função sequenciais. Além disso, mudamos o tipo de resposta: agora a função retorna alguma abstração.

Vamos reescrever o código para funcionar com a versão assíncrona:

def fetch_resource_size(url: str) -> AsyncAbstraction[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

A única coisa diferente é o tipo de retorno - AsyncAbstraction. O resto do código é exatamente o mesmo. Você não precisa mais usar palavras async- chave e await. awaitnão é usado de forma alguma ( por causa disso, tudo foi iniciado ), e sem ele não há sentido em async.

A última coisa é decidir qual cliente precisamos: assíncrono ou síncrono.

def fetch_resource_size(
    client_get: Callable[[str], AbstactionType[Response]],
    url: str,
) -> AbstactionType[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

client_getagora é um argumento de tipo que pode ser chamado que recebe uma string de URL como entrada e retorna algum tipo AbstractionTypesobre o objeto Response. AbstractionType- um Abstractionou AsyncAbstractiondos exemplos anteriores.

Quando passamos Abstraction, o código é executado de forma síncrona, quando AsyncAbstraction- o mesmo código começa a ser executado de forma assíncrona automaticamente.

IOResult e FutureResult

Felizmente, as dry-python/returnsabstrações corretas já existem.

Deixe-me apresentar a você uma ferramenta segura de tipos, compatível com mypy e independente de estruturas, escrita inteiramente em Python. Tem abstrações impressionantes, confortáveis e maravilhosas que podem ser usadas em absolutamente qualquer projeto.

Opção síncrona

Primeiro, vamos adicionar dependências para obter um exemplo reproduzível.

pip install returns httpx anyio

A seguir, vamos transformar o pseudocódigo em código Python funcional. Vamos começar com a opção síncrona.

from typing import Callable
 
import httpx
 
from returns.io import IOResultE, impure_safe
 
def fetch_resource_size(
    client_get: Callable[[str], IOResultE[httpx.Response]],
    url: str,
) -> IOResultE[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )
 
print(fetch_resource_size(
    impure_safe(httpx.get),
    'https://sobolevn.me',
))
# => <IOResult: <Success: 27972>>

Foram necessárias algumas mudanças para que o código funcionasse:

O uso IOResultEé uma forma funcional de lidar com erros síncronos de E / S (as exceções nem sempre funcionam ). Os tipos baseados em Resultpermitem simular exceções, mas com valores separados Failure(). As saídas bem-sucedidas são então agrupadas em um tipo Success. Normalmente ninguém se preocupa com as exceções, mas nós sim.
Use o httpxqual pode lidar com solicitações síncronas e assíncronas.
Use uma função impure_safepara converter o tipo de retorno httpx.getem uma abstração IOResultE.

Opção assíncrona

Vamos tentar fazer o mesmo no código assíncrono.

from typing import Callable
 
import anyio
import httpx
 
from returns.future import FutureResultE, future_safe
 
def fetch_resource_size(
    client_get: Callable[[str], FutureResultE[httpx.Response]],
    url: str,
) -> FutureResultE[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )
 
page_size = fetch_resource_size(
    future_safe(httpx.AsyncClient().get),
    'https://sobolevn.me',
)
print(page_size)
print(anyio.run(page_size.awaitable))
# => <FutureResult: <coroutine object async_map at 0x10b17c320>>
# => <IOResult: <Success: 27972>>

Você vê: o resultado é exatamente o mesmo, mas agora o código está sendo executado de forma assíncrona. No entanto, sua parte principal não mudou. No entanto, você precisa prestar atenção ao seguinte:

Simultaneamente IOResultEalterado para assíncrono FutureResultE, impure_safe- ligado future_safe. Ele funciona da mesma, mas retorna diferente de abstração: FutureResultE.
Usado AsyncClientde httpx.
O valor resultante FutureResultprecisa ser executado porque as funções vermelhas não podem chamar a si mesmas.
O utilitário anyioé usado para mostrar que esta abordagem funciona com qualquer biblioteca assíncrona: asyncio, trio, curio.

Dois em um

Vou mostrar como combinar as versões síncrona e assíncrona em uma API de tipo seguro.

Tipos Kinded superior e de classe tipo para trabalhar com IO ainda não foram liberados (que aparecerá no 0.15.0), por isso vou ilustrar na habitual @overload:

from typing import Callable, Union, overload
 
import anyio
import httpx
 
from returns.future import FutureResultE, future_safe
from returns.io import IOResultE, impure_safe
 
@overload
def fetch_resource_size(
    client_get: Callable[[str], IOResultE[httpx.Response]],
    url: str,
) -> IOResultE[int]:
    """Sync case."""
 
@overload
def fetch_resource_size(
    client_get: Callable[[str], FutureResultE[httpx.Response]],
    url: str,
) -> FutureResultE[int]:
    """Async case."""
 
def fetch_resource_size(
    client_get: Union[
        Callable[[str], IOResultE[httpx.Response]],
        Callable[[str], FutureResultE[httpx.Response]],
    ],
    url: str,
) -> Union[IOResultE[int], FutureResultE[int]]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

Usamos decoradores para @overloaddescrever quais dados de entrada são permitidos e que tipo de valor de retorno será. Você @overloadpode ler mais sobre o decorador em meu outro artigo .

Uma chamada de função com um cliente síncrono ou assíncrono tem a seguinte aparência:

# Sync:
print(fetch_resource_size(
    impure_safe(httpx.get),
    'https://sobolevn.me',
))
# => <IOResult: <Success: 27972>>
 
# Async:
page_size = fetch_resource_size(
    future_safe(httpx.AsyncClient().get),
    'https://sobolevn.me',
)
print(page_size)
print(anyio.run(page_size.awaitable))
# => <FutureResult: <coroutine object async_map at 0x10b17c320>>
# => <IOResult: <Success: 27972>>

Como você pode ver, fetch_resource_sizena variante síncrona, ele retorna IOResulte executa imediatamente . Enquanto na versão assíncrona, um loop de evento é necessário, como para uma co-rotina regular. anyiousado para exibir resultados.

Em mypyeste código não existem comentários:

» mypy async_and_sync.py
Success: no issues found in 1 source file

Vamos ver o que acontece se algo der errado.

---lambda response: len(response.content),
+++lambda response: response.content,

mypy encontra facilmente novos erros:

» mypy async_and_sync.py
async_and_sync.py:33: error: Argument 1 to "map" of "IOResult" has incompatible type "Callable[[Response], bytes]"; expected "Callable[[Response], int]"
async_and_sync.py:33: error: Argument 1 to "map" of "FutureResult" has incompatible type "Callable[[Response], bytes]"; expected "Callable[[Response], int]"
async_and_sync.py:33: error: Incompatible return value type (got "bytes", expected "int")

Prestidigitação e sem mágica: escrever código assíncrono com as abstrações corretas requer apenas a boa e velha composição. Mas o fato de obtermos a mesma API para tipos diferentes é realmente ótimo. Por exemplo, ele permite que você abstraia de como as solicitações HTTP funcionam: de forma síncrona ou assíncrona.

Esperançosamente, este exemplo mostrou como os programas assíncronos podem realmente ser incríveis. E se você tentar dry-python / Returns , encontrará muitas outras coisas interessantes. Na nova versão, já fizemos as primitivas necessárias para trabalhar com Tipos de Tipo Superior e todas as interfaces necessárias. O código acima agora pode ser reescrito assim:

from typing import Callable, TypeVar

import anyio
import httpx

from returns.future import future_safe
from returns.interfaces.specific.ioresult import IOResultLike2
from returns.io import impure_safe
from returns.primitives.hkt import Kind2, kinded

_IOKind = TypeVar('_IOKind', bound=IOResultLike2)

@kinded
def fetch_resource_size(
    client_get: Callable[[str], Kind2[_IOKind, httpx.Response, Exception]],
    url: str,
) -> Kind2[_IOKind, int, Exception]:
    return client_get(url).map(
        lambda response: len(response.content),
    )


# Sync:
print(fetch_resource_size(
    impure_safe(httpx.get),
    'https://sobolevn.me',
))
# => <IOResult: <Success: 27972>>

# Async:
page_size = fetch_resource_size(
    future_safe(httpx.AsyncClient().get),
    'https://sobolevn.me',
)
print(page_size)
print(anyio.run(page_size.awaitable))
# => <FutureResult: <coroutine object async_map at 0x10b17c320>>
# => <IOResult: <Success: 27972>>

Veja o branch `master`, ele já funciona lá.

Mais recursos dry-python

Aqui estão alguns outros recursos úteis do dry-python, dos quais tenho mais orgulho.

Funções digitadas partiale@curry .

from returns.curry import curry, partial
 
def example(a: int, b: str) -> float:
    ...
 
reveal_type(partial(example, 1))
# note: Revealed type is 'def (b: builtins.str) -> builtins.float'
 
reveal_type(curry(example))
# note: Revealed type is 'Overload(def (a: builtins.int) -> def (b: builtins.str) -> builtins.float, def (a: builtins.int, b: builtins.str) -> builtins.float)'

Isso permite que você use @curry, por exemplo, assim:

@curry
def example(a: int, b: str) -> float:
    return float(a + len(b))
 
assert example(1, 'abc') == 4.0
assert example(1)('abc') == 4.0

Pipelines funcionais com inferência de tipo.

Usando um plugin mypy customizado, você pode construir pipelines funcionais que retornam tipos.

from returns.pipeline import flow
assert flow(
    [1, 2, 3],
    lambda collection: max(collection),
    lambda max_number: -max_number,
) == -3

Normalmente em código digitado é muito inconveniente trabalhar com lambdas, pois seus argumentos são sempre do tipo Any. A inferência mypyresolve esse problema.

Com sua ajuda, agora sabemos de que lambda collection: max(collection)tipo Callable[[List[int]], int], mas lambda max_number: -max_numbersimples Callable[[int], int]. O In flowpode transmitir qualquer número de argumentos e eles funcionarão bem. Tudo graças ao plugin.

RequiresContextFutureResultpara injeção de dependência funcional digitada .

A abstração over FutureResult, da qual falamos anteriormente, pode ser usada para passar dependências explicitamente para programas assíncronos em um estilo funcional.

Planos para o futuro

Antes de finalmente lançarmos a versão 1.0, temos que resolver várias tarefas importantes:

Implemente tipos de tipo superior ou sua emulação ( problema ).
Adicione classes de tipo apropriadas para implementar as abstrações necessárias ( problema ).
Talvez tente um compilador mypyc, que irá potencialmente permitir que programas Python anotados com tipo sejam compilados em um binário. Então, o código c dry-python/returnsfuncionará várias vezes mais rápido ( problema ).
Explore novas maneiras de escrever código funcional em Python, como "do-notation" .

conclusões

Composição e abstração podem resolver qualquer problema. Neste artigo, vimos como resolver o problema das cores das funções e escrever um código simples, legível e flexível que funcione. E faça a verificação de tipo.

Experimente dry-python / return e junte-se à Russian Python Week : na conferência, o desenvolvedor de núcleo dry-python Pablo Aguilar realizará um workshop sobre como usar dry-python para escrever lógica de negócios.

O que a assincronia deve ser em Python