👩🏽‍💼 🏂 🖇️ Qué asincronía debería haber en Python 👊🏽 🗼 🍬

En los últimos años, la asyncsemántica y la palabra clave de programación asincrónica han permeado muchos lenguajes de programación populares: JavaScript , Rust , C # y muchos otros . Por supuesto, Python también lo tiene async/await, se introdujo en Python 3.5.

En este artículo quiero discutir los problemas del código asincrónico, especular sobre alternativas y proponer un nuevo enfoque para admitir aplicaciones sincrónicas y asincrónicas al mismo tiempo.

Color de función

Cuando se incluyen funciones asincrónicas en un lenguaje de programación, esencialmente se divide en dos. Aparecen funciones rojas (o asincrónicas) y algunas funciones permanecen azules (sincrónicas).

El principal problema es que las funciones azules no pueden llamar a las rojas, pero las rojas pueden causar potencialmente azules. En Python, por ejemplo, esto es parcialmente cierto: las funciones asincrónicas solo pueden llamar a funciones sincrónicas sin bloqueo. Pero es imposible determinar a partir de la descripción si la función está bloqueando o no. Python es un lenguaje de programación.

Esta división conduce a la división del idioma en dos subconjuntos: sincrónico y asincrónico. Python 3.5 se lanzó hace más de cinco años, pero asynctodavía no es tan compatible como las capacidades sincrónicas de Python.

Puede leer más sobre los colores de función en este excelente artículo .

Código duplicado

Diferentes colores de funciones significan duplicación de código en la práctica.

Imagine que está desarrollando una herramienta CLI para recuperar el tamaño de una página web y desea mantener formas tanto síncronas como asincrónicas de hacerlo. Por ejemplo, esto es necesario si está escribiendo una biblioteca y no sabe cómo se utilizará su código. Y no se trata solo de las bibliotecas PyPI, sino también de nuestras propias bibliotecas con lógica común para varios servicios, escritas, por ejemplo, en Django y aiohttp. Aunque, por supuesto, las aplicaciones independientes se escriben principalmente de forma síncrona o asincrónica.

Comencemos con el pseudocódigo sincrónico:

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

Se ve bien. Ahora veamos el análogo asincrónico:

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

En general, este es el mismo código, pero con la adición de las palabras asyncy await. Y no me lo inventé, compare los ejemplos de código en el tutorial en http:

Hay exactamente la misma imagen.

Abstracción y composición

Resulta que necesita reescribir todo el código sincrónico y organizar aquí y allá asyncy awaitpara que el programa se vuelva asincrónico.

Dos principios pueden ayudar a resolver este problema. Primero, reescribamos el pseudocódigo imperativo en funcional. Esto le permitirá ver la imagen con mayor claridad.

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

Preguntas qué es este método .map, qué hace. Así es como se produce la composición de abstracciones complejas y funciones puras en un estilo funcional. Esto le permite crear una nueva abstracción con un nuevo estado a partir de uno existente. Suponga que client_get(url)inicialmente regresa Abstraction[Response]y la llamada .map(lambda response: len(response.content))convierte la respuesta en la instancia requerida Abstraction[int].

Queda claro qué hacer a continuación. Observe la facilidad con la que pasamos de varios pasos independientes a llamadas de función secuenciales. Además, cambiamos el tipo de respuesta: ahora la función devuelve algo de abstracción.

Reescribamos el código para que funcione con la versión asincrónica:

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

Lo único que es diferente es el tipo de retorno - AsyncAbstraction. El resto del código es exactamente el mismo. Ya no es necesario utilizar palabras clave asyncy await. awaitno se usa en absoluto ( por el bien de esto se inició todo ), y sin él no tiene sentido async.

Lo último es decidir qué cliente necesitamos: asíncrono o 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_getahora es un argumento de tipo invocable que toma una cadena de URL como entrada y devuelve algún tipo AbstractionTypesobre el objeto Response. AbstractionType- cualquiera Abstractiono AsyncAbstractionde los ejemplos anteriores.

Cuando pasamos Abstraction, el código se ejecuta sincrónicamente, cuando AsyncAbstractionel mismo código comienza a ejecutarse automáticamente de forma asincrónica.

IOResult y FutureResult

Afortunadamente, dry-python/returnsya existen las abstracciones correctas.

Permítanme presentarles una herramienta de tipo seguro, compatible con mypy, agnóstica de framework, escrita completamente en Python. Tiene abstracciones asombrosas, cómodas y maravillosas que se pueden usar en absolutamente cualquier proyecto.

Opción sincrónica

Primero, agregaremos dependencias para obtener un ejemplo reproducible.

pip install returns httpx anyio

A continuación, convierta el pseudocódigo en código Python funcional. Comencemos con la opción sincrónica.

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>>

Se necesitaron un par de cambios para que el código funcionara:

El uso IOResultEes una forma funcional de manejar errores de IO síncronos (las excepciones no siempre funcionan ). Los tipos basados en le Resultpermiten simular excepciones, pero con valores separados Failure(). Las salidas exitosas luego se envuelven en un tipo Success. Por lo general, a nadie le importan las excepciones, pero a nosotros sí.
Utilice httpxque pueda manejar solicitudes sincrónicas y asincrónicas.
Utilice una función impure_safepara convertir el tipo de retorno httpx.geten una abstracción IOResultE.

Opción asincrónica

Intentemos hacer lo mismo en código asincrónico.

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>>

Verá: el resultado es exactamente el mismo, pero ahora el código se ejecuta de forma asincrónica. Sin embargo, su parte principal no ha cambiado. Sin embargo, debe prestar atención a lo siguiente:

Simultáneo IOResultEcambiado a asíncrono FutureResultE, impure_safe- activado future_safe. Funciona de la misma, pero devuelve la abstracción diferente: FutureResultE.
Utilizado AsyncClientdesde httpx.
El valor resultante FutureResultdebe ejecutarse porque las funciones rojas no se pueden llamar a sí mismas.
La utilidad anyiose utiliza para mostrar que este enfoque funciona con cualquier biblioteca asíncrona: asyncio, trio, curio.

Dos en uno

Le mostraré cómo combinar las versiones síncrona y asincrónica en una API con seguridad de tipos.

Los tipos de tipo superior y la clase de tipo para trabajar con IO aún no se han lanzado (aparecerán en 0.15.0), por lo que lo ilustraré en lo 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 @overloaddescribir qué datos de entrada están permitidos y qué tipo de valor de retorno será. Puedes leer más sobre el decorador @overloaden mi otro artículo .

Una llamada de función con un cliente sincrónico o asincrónico se ve así:

# 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 puede ver, fetch_resource_sizeen la variante sincrónica regresa inmediatamente IOResulty lo ejecuta. Mientras que en la versión asincrónica, se requiere un bucle de eventos, como para una corrutina regular. anyioutilizado para mostrar resultados.

En mypyeste código no hay comentarios:

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

Veamos qué pasa si algo se estropea.

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

mypy encuentra fácilmente nuevos errores:

» 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")

Juego de manos y sin magia: escribir código asincrónico con las abstracciones correctas requiere simplemente una composición antigua. Pero el hecho de que obtengamos la misma API para diferentes tipos es realmente genial. Por ejemplo, le permite abstraerse de cómo funcionan las solicitudes HTTP: de forma sincrónica o asincrónica.

Con suerte, este ejemplo ha demostrado lo asombrosos que pueden ser los programas asincrónicos. Y si prueba dry-python / returns , encontrará muchas más cosas interesantes. En la nueva versión, ya hemos realizado las primitivas necesarias para trabajar con Higher Kinded Types y todas las interfaces necesarias. El código anterior ahora se puede reescribir así:

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>>

Vea la rama `master`, ya funciona allí.

Más características de dry-python

Aquí hay algunas otras características útiles de dry-python de las que estoy más orgulloso.

Funciones escritas partialy@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)'

Esto le permite usar @curry, por ejemplo, así:

@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 funcionales con inferencia de tipos.

Con un complemento mypy personalizado, puede crear canalizaciones funcionales que devuelvan tipos.

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

Por lo general, es muy inconveniente trabajar con lambdas en código escrito porque sus argumentos son siempre de tipo Any. La inferencia mypyresuelve este problema.

Con su ayuda, ahora sabemos de qué lambda collection: max(collection)tipo Callable[[List[int]], int], pero lambda max_number: -max_numbersimple Callable[[int], int]. In flowpuede pasar cualquier número de argumentos y funcionarán bien. Todo gracias al complemento.

RequiresContextFutureResultpara la inyección de dependencia funcional mecanografiada .

La abstracción over FutureResult, de la que hablamos anteriormente, se puede usar para pasar dependencias explícitamente a programas asincrónicos en un estilo funcional.

Planes para el futuro

Antes de que finalmente lancemos la versión 1.0, tenemos varias tareas importantes que resolver:

Implementar tipos de tipo superior o su emulación ( problema ).
Agregue las clases de tipos adecuadas para implementar las abstracciones requeridas ( problema ).
Tal vez intente un compilador mypyc, que potencialmente permitirá que los programas de Python anotados y escritos se compilen en un binario. Entonces, el código c dry-python/returnsfuncionará varias veces más rápido ( problema ).
Explore nuevas formas de escribir código funcional en Python, como "notación do" .

conclusiones

La composición y la abstracción pueden solucionar cualquier problema. En este artículo, analizamos cómo resolver el problema de los colores de las funciones y escribir un código simple, legible y flexible que funcione. Y verifique el tipo.

Pruebe dry-python / returns y únase a la Russian Python Week : en la conferencia, el desarrollador central de dry-python, Pablo Aguilar, llevará a cabo un taller sobre el uso de dry-python para escribir lógica empresarial.

Qué asincronía debería haber en Python