returns
Version 0.24.0
¡Haga que sus funciones devuelvan algo significativo, escrito y seguro!
mypy , compatible con PEP561¡Inicio rápido ahora mismo!
pip install returns También puede instalar returns con la última versión mypy compatible:
pip install returns[compatible-mypy] También necesitarás configurar nuestro complemento mypy :
# In setup.cfg or mypy.ini:
[mypy]
plugins =
returns.contrib.mypy.returns_plugino:
[ tool . mypy ]
plugins = [ " returns.contrib.mypy.returns_plugin " ] También recomendamos utilizar la misma configuración mypy que utilizamos.
Asegúrate de saber cómo empezar. ¡Consulta nuestros documentos! Pruebe nuestra demostración.
Noneasyncdo-notation para facilitar tu código! None se considera el peor error en la historia de la informática.
Entonces, ¿qué podemos hacer para comprobar que None en nuestros programas? Puede usar el tipo opcional incorporado y escribir muchas condiciones if some is not None: :. Pero tener controles null aquí y allá hace que su código sea ilegible .
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 ) ¡O puedes usar el contenedor Maybe! Consta de los tipos Some y Nothing , que representan el estado existente y el estado vacío (en lugar de None ) respectivamente.
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 Puede estar seguro de que no se llamará al método .bind_optional() para Nothing . ¡Olvídate de los errores relacionados con None para siempre!
También podemos vincular una función de retorno Optional sobre un contenedor. Para lograr esto, usaremos el método .bind_optional .
Y así es como se verá su código refactorizado inicial:
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 ,
)Mucho mejor, ¿no?
Muchos desarrolladores utilizan algún tipo de inyección de dependencia en Python. Y normalmente se basa en la idea de que existe algún tipo de contenedor y proceso de ensamblaje.
¡El enfoque funcional es mucho más simple!
Imagina que tienes un juego basado en django , donde premias a los usuarios con puntos por cada letra adivinada en una palabra (las letras no adivinadas se marcan como '.' ):
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!¡Impresionante! Funciona, los usuarios están contentos, tu lógica es pura e increíble. Pero luego decides hacer el juego más divertido: hagamos que el umbral mínimo de letras responsables sea configurable para un desafío adicional.
Puedes hacerlo directamente:
def _award_points_for_letters ( guessed : int , threshold : int ) -> int :
return 0 if guessed < threshold else guessed El problema es que _award_points_for_letters está profundamente anidado. Y luego tienes que pasar threshold a través de toda la pila de llamadas, incluidos calculate_points y todas las demás funciones que puedan estar en camino. ¡Todos tendrán que aceptar threshold como parámetro! ¡Esto no sirve para nada! Las bases de código grandes tendrán muchas dificultades con este cambio.
Ok, puedes usar directamente django.settings (o similar) en tu función _award_points_for_letters . Y arruina tu lógica pura con detalles específicos del marco . ¡Eso es feo!
O puede utilizar el contenedor RequiresContext . Veamos cómo cambia nuestro código:
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 ,
) Y ahora puedes pasar tus dependencias de una manera realmente directa y explícita. Y tener el tipo de seguridad para comprobar lo que pasa para cubrir su espalda. Consulte los documentos de RequiresContext para obtener más información. Allí aprenderás a hacer '.' También configurable.
También tenemos RequiresContextResult para operaciones relacionadas con el contexto que pueden fallar. Y también RequiresContextIOResult y RequiresContextFutureResult.
Por favor, asegúrese de conocer también la Programación Orientada al Ferrocarril.
Considere este código que puede encontrar en cualquier proyecto 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 () Parece legítimo, ¿no? También parece un código bastante sencillo de probar. Todo lo que necesitas es simular requests.get Obtener para devolver la estructura que necesitas.
Pero hay problemas ocultos en este pequeño código de muestra que son casi imposibles de detectar a primera vista.
Echemos un vistazo al mismo código exacto, pero con todos los problemas ocultos explicados.
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 ()Ahora todos los problemas (¿probablemente todos?) están claros. ¿Cómo podemos estar seguros de que esta función será segura de utilizar dentro de nuestra compleja lógica empresarial?
¡Realmente no podemos estar seguros! Tendremos que crear muchos casos try y except solo para detectar las excepciones esperadas. ¡Nuestro código se volverá complejo e ilegible con todo este lío!
O podemos ir con el nivel superior except Exception: caso para capturar literalmente todo. Y de esta forma acabaríamos pillando a los no deseados. Este enfoque puede ocultarnos problemas graves durante mucho tiempo.
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 ()Ahora tenemos una forma limpia, segura y declarativa de expresar nuestras necesidades comerciales:
Ahora, en lugar de devolver valores regulares, devolvemos valores envueltos dentro de un contenedor especial gracias al decorador @safe. Devolverá Éxito[SuTipo] o Fracaso[Excepción]. ¡Y nunca nos hará ninguna excepción!
También utilizamos funciones de flujo y enlace para una composición práctica y declarativa.
De esta manera podemos estar seguros de que nuestro código no se romperá en lugares aleatorios debido a alguna excepción implícita. Ahora controlamos todas las partes y estamos preparados para errores explícitos.
Aún no hemos terminado con este ejemplo, continuaremos mejorándolo en el próximo capítulo.
Veamos nuestro ejemplo desde otro ángulo. Todas sus funciones parecen normales: es imposible saber si son puras o impuras a primera vista.
Esto lleva a una consecuencia muy importante: comenzamos a mezclar código puro e impuro . ¡No deberíamos hacer eso!
Cuando se mezclan estos dos conceptos sufrimos muchísimo al probarlo o reutilizarlo. Casi todo debería ser puro por defecto. Y deberíamos marcar explícitamente las partes impuras del programa.
Por eso hemos creado un contenedor IO para marcar funciones impuras que nunca fallan.
Estas funciones impuras utilizan fecha y hora random , actual, entorno o consola:
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_numberAhora podemos ver claramente qué funciones son puras y cuáles son impuras. Esto nos ayuda mucho a crear aplicaciones grandes, realizar pruebas unitarias de su código y componer la lógica empresarial juntos.
Como ya se dijo, usamos IO cuando manejamos funciones que no fallan.
¿Qué pasa si nuestra función puede fallar y es impura? Como requests.get() que teníamos anteriormente en nuestro ejemplo.
Entonces tenemos que usar un tipo IOResult especial en lugar de un Result normal. Encontremos la diferencia:
_parse_json siempre devuelve el mismo resultado (con suerte) para la misma entrada: puede analizar json válido o fallar en uno no válido. Es por eso que devolvemos Result puro, no hay IO adentro_make_request es impura y puede fallar. Intenta enviar dos solicitudes similares con y sin conexión a Internet. El resultado será diferente para la misma entrada. Por eso debemos usar IOResult aquí: puede fallar y tiene IOEntonces, para cumplir con nuestro requisito y separar el código puro del impuro, tenemos que refactorizar nuestro ejemplo.
¡Hagamos nuestro IO explícito!
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 ()Y luego podemos usar unsafe_perform_io en algún lugar del nivel superior de nuestro programa para obtener el valor puro (o "real").
Como resultado de esta sesión de refactorización, sabemos todo sobre nuestro código:
Hay varios problemas con el código async en Python:
async desde una sincronizadaawait . ¡Los contenedores Future y FutureResult resuelven estos problemas!
La característica principal de Future es que permite ejecutar código asíncrono manteniendo el contexto de sincronización. Veamos un ejemplo.
Digamos que tenemos dos funciones, la first devuelve un número y la second lo incrementa:
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 intentamos simplemente ejecutar first() , simplemente crearemos una rutina no esperada. No devolverá el valor que queremos.
Pero, si intentáramos ejecutar await first() , entonces necesitaríamos cambiar second para que sea async . Y en ocasiones no es posible por diversos motivos.
Sin embargo, con Future podemos "pretender" llamar al código asíncrono desde el código de sincronización:
from returns . future import Future
def second () -> Future [ int ]:
return Future ( first ()). map ( lambda num : num + 1 ) Sin tocar nuestra first función asíncrona ni hacer second función asíncrona, hemos logrado nuestro objetivo. Ahora, nuestro valor asíncrono se incrementa dentro de una función de sincronización.
Sin embargo, Future todavía requiere ser ejecutado dentro de un bucle de eventos adecuado:
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 Como puede ver, Future le permite trabajar con funciones asíncronas desde un contexto de sincronización. Y mezclar estos dos reinos. Utilice Future sin formato para operaciones que no pueden fallar ni generar excepciones. Prácticamente la misma lógica que teníamos con nuestro contenedor IO .
Ya hemos cubierto cómo funciona Result tanto para código puro como para código impuro. La idea principal es: no planteamos excepciones, las devolvemos. Es especialmente crítico en código asíncrono, porque una sola excepción puede arruinar todas nuestras corrutinas que se ejecutan en un único bucle de eventos.
Tenemos una práctica combinación de contenedores Future y Result : FutureResult . Nuevamente, esto es exactamente como IOResult , pero para código asíncrono impuro. Úselo cuando su Future pueda tener problemas: como solicitudes HTTP u operaciones del sistema de archivos.
Puedes convertir fácilmente cualquier rutina de lanzamiento salvaje en un FutureResult tranquilo:
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 El uso de FutureResult mantendrá su código a salvo de excepciones. Siempre puede await o ejecutar dentro de un bucle de eventos cualquier FutureResult para sincronizar la instancia IOResult y trabajar con él de manera sincronizada.
Anteriormente, tenías que await mucho mientras escribías código 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 )A algunas personas les parece bien, pero a otras no les gusta este estilo imperativo. El problema es que no había otra opción.
¡Pero ahora puedes hacer lo mismo con un estilo funcional! Con la ayuda de los contenedores Future y 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 (...),
)O incluso algo realmente elegante:
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 ),
) Más adelante también podemos refactorizar nuestras funciones lógicas para que se sincronicen y devuelvan FutureResult .
Precioso, ¿no?
¿Quieres más? ¡Vaya a los documentos! O lea estos artículos:
¿Tienes un artículo para enviar? ¡No dudes en abrir una solicitud de extracción!
