Faites en sorte que vos fonctions renvoient quelque chose de significatif, typé et sûr !
mypy
, compatible PEP561Démarrage rapide dès maintenant !
pip install returns
Vous pouvez également installer returns
avec la dernière version mypy
prise en charge :
pip install returns[compatible-mypy]
Vous devrez également configurer notre plugin mypy
:
# In setup.cfg or mypy.ini:
[mypy]
plugins =
returns.contrib.mypy.returns_plugin
ou:
[ tool . mypy ]
plugins = [ " returns.contrib.mypy.returns_plugin " ]
Nous vous recommandons également d'utiliser les mêmes paramètres mypy
que nous utilisons.
Assurez-vous de savoir par où commencer, consultez nos documents ! Essayez notre démo.
None
async
do-notation
pour rendre votre code plus facile ! None
considérée comme la pire erreur de l’histoire de l’informatique.
Alors, que pouvons-nous faire pour vérifier None
dans nos programmes ? Vous pouvez utiliser le type facultatif intégré et écrire beaucoup de conditions if some is not None:
. Mais avoir des vérifications null
ici et là rend votre code illisible .
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 )
Ou vous pouvez utiliser le conteneur Maybe ! Il se compose de types Some
et Nothing
, représentant respectivement l'état existant et l'état vide (au lieu de None
).
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
Vous pouvez être sûr que la méthode .bind_optional()
ne sera pas appelée pour Nothing
. Oubliez pour toujours les erreurs liées à None
!
Nous pouvons également lier une fonction de retour Optional
sur un conteneur. Pour y parvenir, nous allons utiliser la méthode .bind_optional
.
Et voici à quoi ressemblera votre code refactorisé initial :
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 ,
)
C'est beaucoup mieux, n'est-ce pas ?
De nombreux développeurs utilisent une sorte d’injection de dépendances en Python. Et généralement, cela repose sur l’idée qu’il existe une sorte de processus de conteneur et d’assemblage.
L’approche fonctionnelle est beaucoup plus simple !
Imaginez que vous ayez un jeu basé sur django
, dans lequel vous attribuez des points aux utilisateurs pour chaque lettre devinée dans un mot (les lettres non devinées sont marquées par '.'
) :
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!
Génial! Cela fonctionne, les utilisateurs sont satisfaits, votre logique est pure et géniale. Mais plus tard, vous décidez de rendre le jeu plus amusant : rendons le seuil minimal de lettres responsables configurable pour un défi supplémentaire.
Vous pouvez simplement le faire directement :
def _award_points_for_letters ( guessed : int , threshold : int ) -> int :
return 0 if guessed < threshold else guessed
Le problème est que _award_points_for_letters
est profondément imbriqué. Et puis vous devez passer threshold
à travers l'ensemble de la pile d'appels, y compris calculate_points
et toutes les autres fonctions qui pourraient être en route. Tous devront accepter threshold
comme paramètre ! Cela ne sert à rien du tout ! Les grandes bases de code auront beaucoup de difficultés avec ce changement.
Ok, vous pouvez utiliser directement django.settings
(ou similaire) dans votre fonction _award_points_for_letters
. Et ruinez votre logique pure avec des détails spécifiques au framework . C'est moche !
Ou vous pouvez utiliser le conteneur RequiresContext
. Voyons comment notre code change :
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 ,
)
Et maintenant, vous pouvez transmettre vos dépendances de manière vraiment directe et explicite. Et ayez la sécurité de type pour vérifier ce que vous transmettez pour couvrir votre dos. Consultez la documentation RequiresContext pour en savoir plus. Là, vous apprendrez à créer '.'
également configurable.
Nous avons également RequiresContextResult pour les opérations liées au contexte qui pourraient échouer. Et aussi RequiresContextIOResult et RequiresContextFutureResult.
Veuillez vous assurer que vous connaissez également la programmation axée sur les chemins de fer.
Considérez ce code que vous pouvez trouver dans n'importe quel projet python
.
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 ()
Cela semble légitime, n'est-ce pas ? Cela semble également être un code assez simple à tester. Tout ce dont vous avez besoin est de vous moquer requests.get
pour obtenir la structure dont vous avez besoin.
Mais ce petit échantillon de code contient des problèmes cachés qui sont presque impossibles à repérer au premier coup d’œil.
Jetons un coup d'œil au même code exact, mais avec tous les problèmes cachés expliqués.
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 ()
Désormais, tous (probablement tous ?) les problèmes sont clairs. Comment pouvons-nous être sûrs que cette fonction pourra être utilisée en toute sécurité dans notre logique métier complexe ?
Nous ne pouvons vraiment pas en être sûrs ! Nous devrons créer de nombreux cas try
et except
juste pour détecter les exceptions attendues. Notre code va devenir complexe et illisible avec tout ce désordre !
Ou nous pouvons utiliser le niveau supérieur except Exception:
cas pour attraper littéralement tout. Et de cette façon, nous finirions par attraper les indésirables. Cette approche peut nous cacher de graves problèmes pendant longtemps.
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 ()
Nous disposons désormais d’un moyen propre, sûr et déclaratif d’exprimer nos besoins commerciaux :
Désormais, au lieu de renvoyer des valeurs normales, nous renvoyons des valeurs enveloppées dans un conteneur spécial grâce au décorateur @safe. Il renverra Success[YourType] ou Failure[Exception]. Et ne nous fera jamais d’exception !
Nous utilisons également les fonctions flow et bind pour une composition pratique et déclarative.
De cette façon, nous pouvons être sûrs que notre code ne sera pas interrompu à des endroits aléatoires en raison d'une exception implicite. Nous contrôlons désormais toutes les pièces et sommes préparés aux erreurs explicites.
Nous n'en avons pas encore fini avec cet exemple, continuons à l'améliorer dans le chapitre suivant.
Regardons notre exemple sous un autre angle. Toutes ses fonctions semblent normales : il est impossible de dire si elles sont pures ou impures au premier regard.
Cela conduit à une conséquence très importante : nous commençons à mélanger du code pur et impur . Nous ne devrions pas faire ça !
Lorsque ces deux concepts sont mélangés, nous souffrons vraiment lors des tests ou de la réutilisation. Presque tout devrait être pur par défaut. Et nous devrions explicitement marquer les parties impures du programme.
C'est pourquoi nous avons créé un conteneur IO
pour marquer les fonctions impures qui n'échouent jamais.
Ces fonctions impures utilisent random
, la datetime actuelle, l'environnement ou la console :
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_number
Nous pouvons désormais voir clairement quelles fonctions sont pures et lesquelles sont impures. Cela nous aide beaucoup à créer des applications volumineuses, à tester unitairement votre code et à composer ensemble une logique métier.
Comme cela a déjà été dit, nous utilisons IO
lorsque nous traitons des fonctions qui n'échouent pas.
Et si notre fonction pouvait échouer et était impure ? Comme requests.get()
que nous avions plus tôt dans notre exemple.
Ensuite, nous devons utiliser un type IOResult
spécial au lieu d'un Result
normal. Trouvons la différence :
_parse_json
renvoie toujours le même résultat (espérons-le) pour la même entrée : vous pouvez soit analyser json
valide, soit échouer sur un json invalide. C'est pourquoi nous renvoyons Result
pur, il n'y a pas IO
à l'intérieur_make_request
est impure et peut échouer. Essayez d'envoyer deux demandes similaires avec et sans connexion Internet. Le résultat sera différent pour la même entrée. C'est pourquoi nous devons utiliser IOResult
ici : il peut échouer et a IO
Ainsi, afin de répondre à notre exigence et de séparer le code pur du code impur, nous devons refactoriser notre exemple.
Rendons notre IO explicite !
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 ()
Et plus tard, nous pouvons utiliser unsafe_perform_io quelque part au niveau supérieur de notre programme pour obtenir la valeur pure (ou "réelle").
Grâce à cette session de refactoring, nous savons tout de notre code :
Il existe plusieurs problèmes avec le code async
en Python :
async
à partir d'une fonction de synchronisationawait
Les conteneurs Future
et FutureResult
résolvent ces problèmes !
La principale caractéristique de Future est qu'il permet d'exécuter du code asynchrone tout en conservant le contexte de synchronisation. Voyons un exemple.
Disons que nous avons deux fonctions, la first
renvoie un nombre et la second
l'incrémente :
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.
Si nous essayons simplement d'exécuter first()
, nous créerons simplement une coroutine inattendue. Cela ne renverra pas la valeur souhaitée.
Mais si nous essayions d’exécuter await first()
, nous devions alors changer second
pour qu’il soit async
. Et parfois, cela n’est pas possible pour diverses raisons.
Cependant, avec Future
nous pouvons "faire semblant" d'appeler du code asynchrone à partir du code de synchronisation :
from returns . future import Future
def second () -> Future [ int ]:
return Future ( first ()). map ( lambda num : num + 1 )
Sans toucher à notre first
fonction asynchrone ni créer second
fonction asynchrone, nous avons atteint notre objectif. Désormais, notre valeur asynchrone est incrémentée dans une fonction de synchronisation.
Cependant, Future
doit toujours être exécuté dans une boucle d'événement appropriée :
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
Comme vous pouvez le constater, Future
vous permet de travailler avec des fonctions asynchrones à partir d'un contexte de synchronisation. Et de mélanger ces deux domaines. Utilisez Future
brut pour les opérations qui ne peuvent pas échouer ou déclencher des exceptions. À peu près la même logique que celle que nous avions avec notre conteneur IO
.
Nous avons déjà expliqué comment Result
fonctionne pour le code pur et impur. L'idée principale est la suivante : nous ne générons pas d'exceptions, nous les renvoyons. C'est particulièrement critique dans le code asynchrone, car une seule exception peut ruiner toutes nos coroutines exécutées dans une seule boucle d'événement.
Nous avons une combinaison pratique de conteneurs Future
et Result
: FutureResult
. Encore une fois, c'est exactement comme IOResult
, mais pour le code asynchrone impur. Utilisez-le lorsque votre Future
pourrait rencontrer des problèmes : comme des requêtes HTTP ou des opérations sur le système de fichiers.
Vous pouvez facilement transformer n'importe quelle coroutine de lancement sauvage en un FutureResult
calme :
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
L'utilisation de FutureResult
protégera votre code des exceptions. Vous pouvez toujours await
ou exécuter dans une boucle d'événement n'importe quel FutureResult
pour que l'instance de synchronisation IOResult
fonctionne avec elle de manière synchronisée.
Auparavant, vous deviez faire beaucoup d' await
lors de l'écriture de code async
:
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 )
Certaines personnes sont d’accord avec cela, mais d’autres n’aiment pas ce style impératif. Le problème c'est qu'il n'y avait pas le choix.
Mais maintenant, vous pouvez faire la même chose avec un style fonctionnel ! À l'aide des conteneurs Future
et FutureResult
:
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 (...),
)
Ou même quelque chose de vraiment sophistiqué :
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 ),
)
Plus tard, nous pourrons également refactoriser nos fonctions logiques pour qu'elles soient synchronisées et renvoient FutureResult
.
Charmant, n'est-ce pas ?
Vous en voulez plus ? Allez à la documentation ! Ou lisez ces articles :
Avez-vous un article à soumettre ? N'hésitez pas à ouvrir une pull request !