returns
Version 0.24.0
Sorgen Sie dafür, dass Ihre Funktionen etwas Sinnvolles, Typisiertes und Sicheres zurückgeben!
mypy überprüft, PEP561-kompatibelSchnellstart gleich jetzt!
pip install returns Sie können returns auch mit der neuesten unterstützten mypy Version installieren:
pip install returns[compatible-mypy] Sie müssten auch unser mypy -Plugin konfigurieren:
# In setup.cfg or mypy.ini:
[mypy]
plugins =
returns.contrib.mypy.returns_pluginoder:
[ tool . mypy ]
plugins = [ " returns.contrib.mypy.returns_plugin " ] Wir empfehlen außerdem, die gleichen mypy Einstellungen zu verwenden, die wir verwenden.
Stellen Sie sicher, dass Sie wissen, wie Sie beginnen können. Schauen Sie sich unsere Dokumente an! Probieren Sie unsere Demo aus.
None -freien Code schreiben könnenasync Codedo-notation um Ihren Code einfacher zu machen! None wird als der schlimmste Fehler in der Geschichte der Informatik bezeichnet.
Was können wir also tun, um in unseren Programmen nach None zu suchen? Sie können den integrierten optionalen Typ verwenden und viele Bedingungen schreiben if some is not None: . Wenn Sie hier und da jedoch null durchführen, wird Ihr Code unlesbar .
user : Optional [ User ]
discount_program : Optional [ 'DiscountProgram' ] = None
if user is not None :
balance = user . get_balance ()
if balance is not None :
credit = balance . credit_amount ()
if credit is not None and credit > 0 :
discount_program = choose_discount ( credit ) Oder Sie können den Maybe-Container verwenden! Es besteht aus den Typen Some und Nothing , die den vorhandenen Zustand bzw. den leeren Zustand (anstelle von None ) darstellen.
from typing import Optional
from returns . maybe import Maybe , maybe
@ maybe # decorator to convert existing Optional[int] to Maybe[int]
def bad_function () -> Optional [ int ]:
...
maybe_number : Maybe [ float ] = bad_function (). bind_optional (
lambda number : number / 2 ,
)
# => Maybe will return Some[float] only if there's a non-None value
# Otherwise, will return Nothing Sie können sicher sein, dass die Methode .bind_optional() nicht für Nothing aufgerufen wird. Vergessen Sie None -bezogene Fehler für immer!
Wir können auch eine Optional -returning-Funktion über einen Container binden. Um dies zu erreichen, verwenden wir die Methode .bind_optional .
Und so wird Ihr ursprünglich überarbeiteter Code aussehen:
user : Optional [ User ]
# Type hint here is optional, it only helps the reader here:
discount_program : Maybe [ 'DiscountProgram' ] = Maybe . from_optional (
user ,
). bind_optional ( # This won't be called if `user is None`
lambda real_user : real_user . get_balance (),
). bind_optional ( # This won't be called if `real_user.get_balance()` is None
lambda balance : balance . credit_amount (),
). bind_optional ( # And so on!
lambda credit : choose_discount ( credit ) if credit > 0 else None ,
)Viel besser, nicht wahr?
Viele Entwickler verwenden in Python eine Art Abhängigkeitsinjektion. Und normalerweise basiert es auf der Idee, dass es eine Art Container- und Montageprozess gibt.
Der funktionale Ansatz ist viel einfacher!
Stellen Sie sich vor, Sie haben ein django basiertes Spiel, bei dem Sie Benutzern Punkte für jeden erratenen Buchstaben in einem Wort geben (nicht erratene Buchstaben werden mit '.' markiert):
from django . http import HttpRequest , HttpResponse
from words_app . logic import calculate_points
def view ( request : HttpRequest ) -> HttpResponse :
user_word : str = request . POST [ 'word' ] # just an example
points = calculate_points ( user_word )
... # later you show the result to user somehow
# Somewhere in your `words_app/logic.py`:
def calculate_points ( word : str ) -> int :
guessed_letters_count = len ([ letter for letter in word if letter != '.' ])
return _award_points_for_letters ( guessed_letters_count )
def _award_points_for_letters ( guessed : int ) -> int :
return 0 if guessed < 5 else guessed # minimum 6 points possible!Eindrucksvoll! Es funktioniert, die Benutzer sind zufrieden, Ihre Logik ist rein und großartig. Aber später beschließen Sie, das Spiel unterhaltsamer zu gestalten: Lassen Sie uns den Mindestschwellenwert für verantwortliche Buchstaben für eine zusätzliche Herausforderung konfigurierbar machen.
Sie können es einfach direkt tun:
def _award_points_for_letters ( guessed : int , threshold : int ) -> int :
return 0 if guessed < threshold else guessed Das Problem ist, dass _award_points_for_letters tief verschachtelt ist. Und dann müssen Sie threshold durch den gesamten Aufrufstapel übergeben, einschließlich calculate_points und aller anderen Funktionen, die möglicherweise unterwegs sind. Alle müssen threshold als Parameter akzeptieren! Das ist überhaupt nicht sinnvoll! Große Codebasen werden von dieser Änderung stark betroffen sein.
Ok, Sie können django.settings (oder ähnliches) direkt in Ihrer _award_points_for_letters -Funktion verwenden. Und ruinieren Sie Ihre reine Logik mit Framework-spezifischen Details . Das ist hässlich!
Oder Sie können RequiresContext Container verwenden. Mal sehen, wie sich unser Code ändert:
from django . conf import settings
from django . http import HttpRequest , HttpResponse
from words_app . logic import calculate_points
def view ( request : HttpRequest ) -> HttpResponse :
user_word : str = request . POST [ 'word' ] # just an example
points = calculate_points ( user_words )( settings ) # passing the dependencies
... # later you show the result to user somehow
# Somewhere in your `words_app/logic.py`:
from typing import Protocol
from returns . context import RequiresContext
class _Deps ( Protocol ): # we rely on abstractions, not direct values or types
WORD_THRESHOLD : int
def calculate_points ( word : str ) -> RequiresContext [ int , _Deps ]:
guessed_letters_count = len ([ letter for letter in word if letter != '.' ])
return _award_points_for_letters ( guessed_letters_count )
def _award_points_for_letters ( guessed : int ) -> RequiresContext [ int , _Deps ]:
return RequiresContext (
lambda deps : 0 if guessed < deps . WORD_THRESHOLD else guessed ,
) Und jetzt können Sie Ihre Abhängigkeiten wirklich direkt und explizit übergeben. Und verfügen Sie über die Typensicherheit, um zu überprüfen, was Sie passieren, um Ihren Rücken zu schützen. Weitere Informationen finden Sie in der RequiresContext-Dokumentation. Dort erfahren Sie, wie man '.' auch konfigurierbar.
Wir haben auch RequiresContextResult für kontextbezogene Vorgänge, die möglicherweise fehlschlagen. Und auch RequiresContextIOResult und RequiresContextFutureResult.
Bitte stellen Sie sicher, dass Sie auch mit Railway Oriented Programming vertraut sind.
Betrachten Sie diesen Code, den Sie in jedem python -Projekt finden können.
import requests
def fetch_user_profile ( user_id : int ) -> 'UserProfile' :
"""Fetches UserProfile dict from foreign API."""
response = requests . get ( '/api/users/{0}' . format ( user_id ))
response . raise_for_status ()
return response . json () Scheint echt zu sein, nicht wahr? Es scheint auch ein ziemlich einfach zu testender Code zu sein. Sie müssen lediglich requests.get verspotten, um die benötigte Struktur zurückzugeben.
Allerdings gibt es in diesem winzigen Codebeispiel versteckte Probleme, die auf den ersten Blick kaum zu erkennen sind.
Werfen wir einen Blick auf genau denselben Code, jedoch mit Erklärung aller versteckten Probleme.
import requests
def fetch_user_profile ( user_id : int ) -> 'UserProfile' :
"""Fetches UserProfile dict from foreign API."""
response = requests . get ( '/api/users/{0}' . format ( user_id ))
# What if we try to find user that does not exist?
# Or network will go down? Or the server will return 500?
# In this case the next line will fail with an exception.
# We need to handle all possible errors in this function
# and do not return corrupt data to consumers.
response . raise_for_status ()
# What if we have received invalid JSON?
# Next line will raise an exception!
return response . json ()Jetzt sind alle (wahrscheinlich alle?) Probleme klar. Wie können wir sicher sein, dass diese Funktion innerhalb unserer komplexen Geschäftslogik sicher verwendet werden kann?
Wir können wirklich nicht sicher sein! Wir müssen viele try und except -Fälle erstellen, um die erwarteten Ausnahmen abzufangen. Unser Code wird bei all dem Durcheinander komplex und unlesbar!
Oder wir können mit der obersten Ebene except Exception: case fortfahren, um buchstäblich alles zu erfassen. Und auf diese Weise würden wir am Ende unerwünschte Tiere fangen. Dieser Ansatz kann ernsthafte Probleme für lange Zeit vor uns verbergen.
import requests
from returns . result import Result , safe
from returns . pipeline import flow
from returns . pointfree import bind
def fetch_user_profile ( user_id : int ) -> Result [ 'UserProfile' , Exception ]:
"""Fetches `UserProfile` TypedDict from foreign API."""
return flow (
user_id ,
_make_request ,
bind ( _parse_json ),
)
@ safe
def _make_request ( user_id : int ) -> requests . Response :
# TODO: we are not yet done with this example, read more about `IO`:
response = requests . get ( '/api/users/{0}' . format ( user_id ))
response . raise_for_status ()
return response
@ safe
def _parse_json ( response : requests . Response ) -> 'UserProfile' :
return response . json ()Jetzt haben wir eine saubere, sichere und deklarative Möglichkeit, unsere Geschäftsanforderungen auszudrücken:
Anstatt reguläre Werte zurückzugeben, geben wir jetzt dank des @safe-Dekorators Werte zurück, die in einen speziellen Container eingeschlossen sind. Es wird Success[YourType] oder Failure[Exception] zurückgegeben. Und wird uns niemals aus dem Konzept bringen!
Wir verwenden auch Flow- und Bind-Funktionen für eine praktische und deklarative Komposition.
Auf diese Weise können wir sicher sein, dass unser Code nicht aufgrund einer impliziten Ausnahme an zufälligen Stellen abbricht. Jetzt kontrollieren wir alle Teile und sind auf die expliziten Fehler vorbereitet.
Wir sind mit diesem Beispiel noch nicht fertig, wir verbessern es im nächsten Kapitel weiter.
Betrachten wir unser Beispiel aus einem anderen Blickwinkel. Alle seine Funktionen sehen wie normale aus: Ob sie rein oder unrein sind, lässt sich auf den ersten Blick nicht erkennen.
Dies führt zu einer sehr wichtigen Konsequenz: Wir beginnen, reinen und unreinen Code miteinander zu vermischen . Das sollten wir nicht tun!
Wenn diese beiden Konzepte gemischt werden, leiden wir beim Testen oder Wiederverwenden sehr darunter. Fast alles sollte standardmäßig rein sein. Und wir sollten unreine Teile des Programms explizit kennzeichnen.
Aus diesem Grund haben wir IO Container erstellt, um unreine Funktionen zu markieren, die niemals fehlschlagen.
Diese unreinen Funktionen verwenden random , aktuelles Datum/Uhrzeit, Umgebung oder Konsole:
import random
import datetime as dt
from returns . io import IO
def get_random_number () -> IO [ int ]: # or use `@impure` decorator
return IO ( random . randint ( 1 , 10 )) # isn't pure, because random
now : Callable [[], IO [ dt . datetime ]] = impure ( dt . datetime . now )
@ impure
def return_and_show_next_number ( previous : int ) -> int :
next_number = previous + 1
print ( next_number ) # isn't pure, because does IO
return next_numberJetzt können wir klar erkennen, welche Funktionen rein und welche unrein sind. Dies hilft uns sehr beim Erstellen großer Anwendungen, beim Unit-Testen Ihres Codes und beim gemeinsamen Verfassen von Geschäftslogik.
Wie bereits erwähnt, verwenden wir IO , wenn wir Funktionen verarbeiten, die nicht fehlschlagen.
Was ist, wenn unsere Funktion versagen kann und unrein ist? Wie requests.get() das wir zuvor in unserem Beispiel hatten.
Dann müssen wir einen speziellen IOResult Typ anstelle eines regulären Result verwenden. Finden wir den Unterschied:
_parse_json -Funktion gibt (hoffentlich) immer das gleiche Ergebnis für die gleiche Eingabe zurück: Sie können entweder gültiges json analysieren oder bei ungültigem fehlschlagen. Deshalb geben wir reines Result zurück, es gibt keine IO darin_make_request ist unrein und kann fehlschlagen. Versuchen Sie, zwei ähnliche Anfragen mit und ohne Internetverbindung zu senden. Das Ergebnis wird für die gleiche Eingabe unterschiedlich sein. Deshalb müssen wir hier IOResult verwenden: Es kann fehlschlagen und verfügt über IOUm unsere Anforderung zu erfüllen und reinen Code von unreinem zu trennen, müssen wir unser Beispiel umgestalten.
Machen wir unser IO explizit!
import requests
from returns . io import IOResult , impure_safe
from returns . result import safe
from returns . pipeline import flow
from returns . pointfree import bind_result
def fetch_user_profile ( user_id : int ) -> IOResult [ 'UserProfile' , Exception ]:
"""Fetches `UserProfile` TypedDict from foreign API."""
return flow (
user_id ,
_make_request ,
# before: def (Response) -> UserProfile
# after safe: def (Response) -> ResultE[UserProfile]
# after bind_result: def (IOResultE[Response]) -> IOResultE[UserProfile]
bind_result ( _parse_json ),
)
@ impure_safe
def _make_request ( user_id : int ) -> requests . Response :
response = requests . get ( '/api/users/{0}' . format ( user_id ))
response . raise_for_status ()
return response
@ safe
def _parse_json ( response : requests . Response ) -> 'UserProfile' :
return response . json ()Und später können wir unsafe_perform_io irgendwo auf der obersten Ebene unseres Programms verwenden, um den reinen (oder „echten“) Wert zu erhalten.
Als Ergebnis dieser Refactoring-Sitzung wissen wir alles über unseren Code:
Es gibt mehrere Probleme mit async Code in Python:
async Funktion nicht von einer synchronisierten aus aufrufenawait -Anweisungen Future und FutureResult Container lösen diese Probleme!
Das Hauptmerkmal von Future besteht darin, dass es die Ausführung von asynchronem Code unter Beibehaltung des Synchronisierungskontexts ermöglicht. Sehen wir uns ein Beispiel an.
Nehmen wir an, wir haben zwei Funktionen, die first gibt eine Zahl zurück und die second erhöht sie:
async def first () -> int :
return 1
def second (): # How can we call `first()` from here?
return first () + 1 # Boom! Don't do this. We illustrate a problem here. Wenn wir versuchen, einfach first() auszuführen, erstellen wir einfach eine unerwartete Coroutine. Es wird nicht den gewünschten Wert zurückgeben.
Wenn wir jedoch versuchen würden, await first() auszuführen, müssten wir second in async “ ändern. Und manchmal ist es aus verschiedenen Gründen nicht möglich.
Allerdings können wir mit Future „so tun“, als würden wir asynchronen Code aus synchronem Code aufrufen:
from returns . future import Future
def second () -> Future [ int ]:
return Future ( first ()). map ( lambda num : num + 1 ) Ohne unsere first asynchrone Funktion zu berühren oder second asynchrone Funktion zu erstellen, haben wir unser Ziel erreicht. Jetzt wird unser asynchroner Wert innerhalb einer Synchronisierungsfunktion erhöht.
Future muss jedoch weiterhin in einer richtigen Ereignisschleife ausgeführt werden:
import anyio # or asyncio, or any other lib
# We can then pass our `Future` to any library: asyncio, trio, curio.
# And use any event loop: regular, uvloop, even a custom one, etc
assert anyio . run ( second (). awaitable ) == 2 Wie Sie sehen, ermöglicht Ihnen Future die Arbeit mit asynchronen Funktionen aus einem Synchronisierungskontext. Und diese beiden Bereiche miteinander zu vermischen. Verwenden Sie Raw Future für Vorgänge, die nicht fehlschlagen oder Ausnahmen auslösen können. So ziemlich die gleiche Logik, die wir bei unserem IO Container hatten.
Wir haben bereits erläutert, wie Result sowohl für reinen als auch für unreinen Code funktioniert. Die Grundidee ist: Wir lösen keine Ausnahmen aus, wir geben sie zurück. Dies ist besonders wichtig bei asynchronem Code, da eine einzige Ausnahme alle unsere Coroutinen ruinieren kann, die in einer einzigen Ereignisschleife ausgeführt werden.
Wir haben eine praktische Kombination aus Future und Result Containern: FutureResult . Auch dies ist genau wie IOResult , jedoch für unreinen asynchronen Code. Verwenden Sie es, wenn Ihre Future möglicherweise Probleme hat: wie HTTP-Anfragen oder Dateisystemvorgänge.
Sie können jede wild werfende Coroutine ganz einfach in ein ruhiges FutureResult verwandeln:
import anyio
from returns . future import future_safe
from returns . io import IOFailure
@ future_safe
async def raising ():
raise ValueError ( 'Not so fast!' )
ioresult = anyio . run ( raising . awaitable ) # all `Future`s return IO containers
assert ioresult == IOFailure ( ValueError ( 'Not so fast!' )) # True Durch die Verwendung von FutureResult wird Ihr Code vor Ausnahmen geschützt. Sie können jedes FutureResult jederzeit in einer Ereignisschleife await oder ausführen, um eine synchronisierte IOResult Instanz zu erhalten, die synchron damit arbeitet.
Bisher musste man beim Schreiben von async Code ziemlich lange await :
async def fetch_user ( user_id : int ) -> 'User' :
...
async def get_user_permissions ( user : 'User' ) -> 'Permissions' :
...
async def ensure_allowed ( permissions : 'Permissions' ) -> bool :
...
async def main ( user_id : int ) -> bool :
# Also, don't forget to handle all possible errors with `try / except`!
user = await fetch_user ( user_id ) # We will await each time we use a coro!
permissions = await get_user_permissions ( user )
return await ensure_allowed ( permissions )Manche Leute sind damit einverstanden, aber manche mögen diesen zwingenden Stil nicht. Das Problem ist, dass es keine Wahl gab.
Aber jetzt können Sie dasselbe im funktionalen Stil tun! Mit Hilfe von Future und FutureResult -Containern:
import anyio
from returns . future import FutureResultE , future_safe
from returns . io import IOSuccess , IOFailure
@ future_safe
async def fetch_user ( user_id : int ) -> 'User' :
...
@ future_safe
async def get_user_permissions ( user : 'User' ) -> 'Permissions' :
...
@ future_safe
async def ensure_allowed ( permissions : 'Permissions' ) -> bool :
...
def main ( user_id : int ) -> FutureResultE [ bool ]:
# We can now turn `main` into a sync function, it does not `await` at all.
# We also don't care about exceptions anymore, they are already handled.
return fetch_user ( user_id ). bind ( get_user_permissions ). bind ( ensure_allowed )
correct_user_id : int # has required permissions
banned_user_id : int # does not have required permissions
wrong_user_id : int # does not exist
# We can have correct business results:
assert anyio . run ( main ( correct_user_id ). awaitable ) == IOSuccess ( True )
assert anyio . run ( main ( banned_user_id ). awaitable ) == IOSuccess ( False )
# Or we can have errors along the way:
assert anyio . run ( main ( wrong_user_id ). awaitable ) == IOFailure (
UserDoesNotExistError (...),
)Oder sogar etwas ganz Besonderes:
from returns . pointfree import bind
from returns . pipeline import flow
def main ( user_id : int ) -> FutureResultE [ bool ]:
return flow (
fetch_user ( user_id ),
bind ( get_user_permissions ),
bind ( ensure_allowed ),
) Später können wir unsere logischen Funktionen auch so umgestalten, dass sie synchronisiert sind und FutureResult zurückgeben.
Schön, nicht wahr?
Lust auf mehr? Gehen Sie zu den Dokumenten! Oder lesen Sie diese Artikel:
Möchten Sie einen Artikel einreichen? Eröffnen Sie gerne eine Pull-Anfrage!
