➖ 🐡 👡 Welche Asynchronität sollte in Python sein? 🖖🏾 🔂 🕣

In den letzten Jahren haben das Schlüsselwort asyncund die Semantik der asynchronen Programmierung viele beliebte Programmiersprachen durchdrungen: JavaScript , Rust , C # und viele andere . Natürlich hat Python es auch async/await, es wurde in Python 3.5 eingeführt.

In diesem Artikel möchte ich die Probleme des asynchronen Codes diskutieren, über Alternativen spekulieren und einen neuen Ansatz vorschlagen, um sowohl synchrone als auch asynchrone Anwendungen gleichzeitig zu unterstützen.

Funktionsfarbe

Wenn asynchrone Funktionen in einer Programmiersprache enthalten sind, wird sie im Wesentlichen in zwei Teile geteilt. Rote Funktionen werden angezeigt (oder asynchron) und einige Funktionen bleiben blau (synchron).

Das Hauptproblem besteht darin, dass blaue Funktionen keine roten Funktionen aufrufen können, rote jedoch möglicherweise blaue Funktionen verursachen können. In Python ist dies beispielsweise teilweise der Fall: Asynchrone Funktionen können nur synchrone nicht blockierende Funktionen aufrufen. Anhand der Beschreibung kann jedoch nicht festgestellt werden, ob die Funktion blockiert oder nicht. Python ist eine Skriptsprache.

Diese Aufteilung führt zur Aufteilung der Sprache in zwei Teilmengen: synchron und asynchron. Python 3.5 wurde vor über fünf Jahren veröffentlicht, wird jedoch asyncbei weitem nicht so gut unterstützt wie die synchronen Funktionen von Python.

Weitere Informationen zu Funktionsfarben finden Sie in diesem großartigen Artikel .

Doppelter Code

Unterschiedliche Farben von Funktionen bedeuten in der Praxis eine Codeduplizierung.

Stellen Sie sich vor, Sie entwickeln ein CLI-Tool zum Abrufen der Größe einer Webseite und möchten sowohl synchrone als auch asynchrone Methoden beibehalten. Dies ist beispielsweise erforderlich, wenn Sie eine Bibliothek schreiben und nicht wissen, wie Ihr Code verwendet wird. Und es geht nicht nur um PyPI-Bibliotheken, sondern auch um unsere eigenen Bibliotheken mit gemeinsamer Logik für verschiedene Dienste, die beispielsweise in Django und aiohttp geschrieben sind. Obwohl unabhängige Anwendungen natürlich meistens entweder nur synchron oder nur asynchron geschrieben werden.

Beginnen wir mit dem synchronen Pseudocode:

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

Sieht gut aus. Schauen wir uns nun das asynchrone Analogon an:

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

Im Allgemeinen ist dies der gleiche Code, jedoch mit den Wörtern asyncund await. Und ich habe es nicht erfunden - vergleichen Sie die Codebeispiele im Tutorial auf httpx:

synchroner Code ;
asynchroner Code .

Es gibt genau das gleiche Bild.

Abstraktion und Zusammensetzung

Es stellt sich heraus , dass Sie alle Synchroncode und ordnen hier und dort neu zu schreiben müssen asyncund awaitso , dass das Programm asynchron wird.

Zwei Prinzipien können helfen, dieses Problem zu lösen. Lassen Sie uns zunächst den imperativen Pseudocode in funktional umschreiben. Auf diese Weise können Sie das Bild klarer sehen.

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

Sie fragen, was diese Methode ist .map, was sie tut. So erfolgt die Komposition komplexer Abstraktionen und reiner Funktionen in einem funktionalen Stil. Auf diese Weise können Sie eine neue Abstraktion mit einem neuen Status aus einem vorhandenen erstellen. Angenommen, es wird client_get(url)zunächst zurückgegeben Abstraction[Response]und der Aufruf .map(lambda response: len(response.content))konvertiert die Antwort in die erforderliche Instanz Abstraction[int].

Es wird klar, was als nächstes zu tun ist. Beachten Sie, wie einfach wir von einigen unabhängigen Schritten zu sequentiellen Funktionsaufrufen übergegangen sind. Außerdem haben wir den Antworttyp geändert: Jetzt gibt die Funktion eine gewisse Abstraktion zurück.

Schreiben wir den Code neu, um mit der asynchronen Version zu arbeiten:

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

Das einzige, was anders ist, ist der Rückgabetyp - AsyncAbstraction. Der Rest des Codes ist genau der gleiche. Sie müssen keine Schlüsselwörter asyncund mehr verwenden await. awaitwird überhaupt nicht verwendet ( aus diesem Grund wurde alles gestartet ), und ohne es macht es keinen Sinn async.

Als letztes müssen Sie entscheiden, welchen Client wir benötigen: asynchron oder synchron.

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_getist jetzt ein aufrufbares Typargument, das eine URL-Zeichenfolge als Eingabe verwendet und einen Typ AbstractionTypeüber das Objekt zurückgibt Response. AbstractionType- entweder Abstractionoder AsyncAbstractionaus den vorherigen Beispielen.

Wenn wir übergeben Abstraction, wird der Code synchron ausgeführt, wenn AsyncAbstraction- derselbe Code automatisch asynchron ausgeführt wird.

IOResult und FutureResult

Glücklicherweise sind die dry-python/returnsrichtigen Abstraktionen bereits vorhanden.

Lassen Sie mich Ihnen ein typsicheres, mypy-freundliches, rahmenunabhängiges, vollständig Python-Tool vorstellen. Es hat erstaunliche, handliche, wundervolle Abstraktionen, die in absolut jedem Projekt verwendet werden können.

Synchrone Option

Fügen wir zunächst Abhängigkeiten hinzu, um ein reproduzierbares Beispiel zu erhalten

pip install returns httpx anyio

Als nächstes verwandeln wir den Pseudocode in funktionierenden Python-Code. Beginnen wir mit der synchronen Option.

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

Es waren einige Änderungen erforderlich, um einen funktionierenden Code zu erhalten:

Die Verwendung IOResultEist eine funktionale Methode zur Behandlung synchroner E / A-Fehler ( Ausnahmen funktionieren nicht immer ). Mit Typen, die auf basieren Result, können Sie Ausnahmen simulieren, jedoch mit separaten Werten Failure(). Erfolgreiche Exits werden dann in einen Typ eingeschlossen Success. Normalerweise kümmert sich niemand um Ausnahmen, aber wir tun es.
Verwendung, httpxdie synchrone und asynchrone Anforderungen verarbeiten kann.
Verwenden Sie eine Funktion impure_safe, um den Rückgabetyp httpx.getin eine Abstraktion zu konvertieren IOResultE.

Asynchrone Option

Versuchen wir, dasselbe in asynchronem Code zu tun.

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

Sie sehen: Das Ergebnis ist genau das gleiche, aber jetzt wird der Code asynchron ausgeführt. Der Hauptteil hat sich jedoch nicht geändert. Sie müssen jedoch Folgendes beachten:

Gleichzeitig IOResultEgeändert in asynchron FutureResultE, impure_safe- ein future_safe. Es funktioniert genauso, gibt aber die unterschiedliche Abstraktion zurück : FutureResultE.
Verwendet AsyncClientvon httpx.
Der resultierende Wert FutureResultmuss ausgeführt werden, da sich rote Funktionen nicht selbst aufrufen können.
Das Dienstprogramm anyiowird verwendet , um zu zeigen , dass dieser Ansatz mit einer beliebigen asynchronen Bibliothek funktioniert: asyncio, trio, curio.

Zwei in eins

Ich zeige Ihnen, wie Sie die synchronen und asynchronen Versionen in einer typsicheren API kombinieren.

Höher sortierte Typen und eine Typklasse für die Arbeit mit E / A wurden noch nicht veröffentlicht (sie werden in 0.15.0 erscheinen), daher werde ich Folgendes veranschaulichen @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),
    )

Wir verwenden Dekoratoren, um zu @overloadbeschreiben, welche Eingabedaten zulässig sind und welche Art von Rückgabewert sein wird. Sie @overloadkönnen mehr über den Dekorateur in meinem anderen Artikel lesen .

Ein Funktionsaufruf mit einem synchronen oder asynchronen Client sieht folgendermaßen aus:

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

Wie Sie sehen können, wird es fetch_resource_sizein der synchronen Version sofort zurückgegeben IOResultund ausgeführt. Während in der asynchronen Version eine Ereignisschleife erforderlich ist, wie bei einer regulären Coroutine. anyiowird verwendet, um Ergebnisse anzuzeigen.

In mypydiesem Code gibt es keine Kommentare:

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

Mal sehen, was passiert, wenn etwas durcheinander ist.

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

mypy findet leicht neue Fehler:

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

Fingerspitzengefühl und keine Magie: Das Schreiben von asynchronem Code mit den richtigen Abstraktionen erfordert nur eine gute, altmodische Komposition. Aber die Tatsache, dass wir die gleiche API für verschiedene Typen erhalten, ist wirklich großartig. So können Sie beispielsweise abstrahieren, wie HTTP-Anforderungen funktionieren: synchron oder asynchron.

Hoffentlich hat dieses Beispiel gezeigt, wie großartig asynchrone Programme wirklich sein können. Und wenn Sie Dry-Python / Returns ausprobieren , werden Sie viele weitere interessante Dinge finden. In der neuen Version haben wir bereits die erforderlichen Grundelemente für die Arbeit mit höherwertigen Typen und alle erforderlichen Schnittstellen erstellt. Der obige Code kann jetzt folgendermaßen umgeschrieben werden:

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

Siehe den Zweig "Master", der dort bereits funktioniert.

Weitere Dry-Python-Funktionen

Hier sind einige andere nützliche Dry-Python-Funktionen, auf die ich besonders stolz bin.

Eingegebene Funktionen partialund@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)'

Auf diese Weise können Sie @currybeispielsweise Folgendes verwenden:

@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

Funktionspipelines mit Typinferenz.

Mit einem benutzerdefinierten mypy-Plugin können Sie funktionale Pipelines erstellen, die Typen zurückgeben.

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

Es ist normalerweise sehr unpraktisch, mit Lambdas in typisiertem Code zu arbeiten, da ihre Argumente immer vom Typ sind Any. Die Folgerung mypylöst dieses Problem.

Mit seiner Hilfe wissen wir jetzt, welcher lambda collection: max(collection)Typ Callable[[List[int]], int], aber lambda max_number: -max_numbereinfach Callable[[int], int]. In flowkann eine beliebige Anzahl von Argumenten übergeben, und sie funktionieren einwandfrei. Alles dank dem Plugin.

RequiresContextFutureResultfür typisierte funktionale Abhängigkeitsinjektion .

Die Abstraktion über FutureResult, über die wir zuvor gesprochen haben, kann verwendet werden, um Abhängigkeiten in einem funktionalen Stil explizit an asynchrone Programme zu übergeben.

Pläne für die Zukunft

Bevor wir endlich Version 1.0 veröffentlichen, müssen wir einige wichtige Aufgaben lösen:

Implementieren Sie höherwertige Typen oder deren Emulation ( Problem ).
Fügen Sie geeignete Typklassen hinzu, um die erforderlichen Abstraktionen zu implementieren ( Problem ).
Versuchen Sie es mit einem Compiler mypyc, mit dem typisierte kommentierte Python-Programme möglicherweise in eine Binärdatei kompiliert werden können. Dann dry-python/returnsfunktioniert der Code mit mehrmals schneller ( Problem ).
Entdecken Sie neue Möglichkeiten zum Schreiben von Funktionscode in Python, z. B. "Do-Notation" .

Schlussfolgerungen

Komposition und Abstraktion können jedes Problem lösen. In diesem Artikel haben wir uns angesehen, wie Sie das Problem der Funktionsfarben lösen und einfachen, lesbaren und flexiblen Code schreiben können, der funktioniert. Und machen Sie eine Typprüfung.

Probieren Sie Dry-Python / Returns aus und nehmen Sie an der Russian Python Week teil : Auf der Konferenz wird der Dry-Python-Kernentwickler Pablo Aguilar einen Workshop über die Verwendung von Dry-Python zum Schreiben von Geschäftslogik abhalten .

Welche Asynchronität sollte in Python sein?