returns
Version 0.24.0
Faça com que suas funções retornem algo significativo, digitado e seguro!
mypy , compatível com PEP561Início rápido agora mesmo!
pip install returns Você também pode instalar returns com a versão mypy mais recente suportada:
pip install returns[compatible-mypy] Você também precisaria configurar nosso plugin mypy :
# In setup.cfg or mypy.ini:
[mypy]
plugins =
returns.contrib.mypy.returns_pluginou:
[ tool . mypy ]
plugins = [ " returns.contrib.mypy.returns_plugin " ] Também recomendamos usar as mesmas configurações mypy que usamos.
Certifique-se de saber como começar, confira nossos documentos! Experimente nossa demonstração.
None -freeasyncdo-notation para tornar seu código mais fácil! None é considerado o pior erro da história da Ciência da Computação.
Então, o que podemos fazer para verificar None em nossos programas? Você pode usar o tipo opcional integrado e escrever muitas condições if some is not None: :. Mas ter verificações null aqui e ali torna seu código ilegível .
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 você pode usar o contêiner Maybe! Consiste nos tipos Some e Nothing , representando o estado existente e o estado vazio (em vez 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 Você pode ter certeza de que o método .bind_optional() não será chamado para Nothing . Esqueça os erros relacionados None para sempre!
Também podemos vincular uma função de retorno Optional a um contêiner. Para conseguir isso, usaremos o método .bind_optional .
E aqui está a aparência do seu código refatorado 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 ,
)Muito melhor, não é?
Muitos desenvolvedores usam algum tipo de injeção de dependência em Python. E geralmente é baseado na ideia de que existe algum tipo de contêiner e processo de montagem.
A abordagem funcional é muito mais simples!
Imagine que você tem um jogo baseado em django , onde você premia os usuários com pontos para cada letra adivinhada em uma palavra (letras não adivinhadas são marcadas 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!Incrível! Funciona, os usuários estão felizes, sua lógica é pura e incrível. Mas, mais tarde, você decide tornar o jogo mais divertido: vamos tornar o limite mínimo de letras responsáveis configurável para um desafio extra.
Você pode fazer isso diretamente:
def _award_points_for_letters ( guessed : int , threshold : int ) -> int :
return 0 if guessed < threshold else guessed O problema é que _award_points_for_letters está profundamente aninhado. E então você tem que passar threshold por toda a pilha de chamadas, incluindo calculate_points e todas as outras funções que possam estar a caminho. Todos eles terão que aceitar threshold como parâmetro! Isso não é nada útil! Grandes bases de código sofrerão muito com essa mudança.
Ok, você pode usar diretamente django.settings (ou similar) em sua função _award_points_for_letters . E estrague sua lógica pura com detalhes específicos do framework . Isso é feio!
Ou você pode usar o contêiner RequiresContext . Vamos ver como nosso código muda:
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 ,
) E agora você pode passar suas dependências de uma forma bem direta e explícita. E tenha segurança de tipo para verificar o que você passa para proteger suas costas. Confira a documentação do RequiresContext para mais informações. Lá você aprenderá como fazer '.' também configurável.
Também temos RequiresContextResult para operações relacionadas ao contexto que podem falhar. E também RequiresContextIOResult e RequiresContextFutureResult.
Por favor, certifique-se de que você também conhece a Programação Orientada para Ferrovias.
Considere este código que você pode encontrar em qualquer projeto 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, não é? Também parece um código bastante simples de testar. Tudo que você precisa é simular requests.get para retornar a estrutura necessária.
Porém, existem problemas ocultos neste pequeno exemplo de código que são quase impossíveis de detectar à primeira vista.
Vamos dar uma olhada exatamente no mesmo código, mas com todos os 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 ()Agora, todos (provavelmente todos?) os problemas estão claros. Como podemos ter certeza de que essa função será segura para uso dentro de nossa complexa lógica de negócios?
Realmente não podemos ter certeza! Teremos que criar muitos casos try e except apenas para capturar as exceções esperadas. Nosso código ficará complexo e ilegível com toda essa bagunça!
Ou podemos ir com o nível superior except Exception: case para capturar literalmente tudo. E assim acabaríamos pegando os indesejados. Essa abordagem pode esconder de nós problemas sérios por muito tempo.
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 ()Agora temos uma forma limpa, segura e declarativa de expressar nossas necessidades de negócios:
Agora, em vez de retornar valores regulares, retornamos valores agrupados em um contêiner especial graças ao decorador @safe. Ele retornará Success[YourType] ou Failure[Exception]. E nunca lançará exceção para nós!
Também usamos funções de fluxo e ligação para composição prática e declarativa.
Dessa forma podemos ter certeza de que nosso código não quebrará em locais aleatórios devido a alguma exceção implícita. Agora controlamos todas as partes e estamos preparados para erros explícitos.
Ainda não terminamos este exemplo, vamos continuar a melhorá-lo no próximo capítulo.
Vejamos nosso exemplo de outro ângulo. Todas as suas funções parecem normais: é impossível dizer à primeira vista se são puras ou impuras.
Isso leva a uma consequência muito importante: começamos a misturar código puro e impuro . Não deveríamos fazer isso!
Quando esses dois conceitos se misturam sofremos muito na hora de testá-lo ou reutilizá-lo. Quase tudo deveria ser puro por padrão. E deveríamos marcar explicitamente as partes impuras do programa.
É por isso que criamos o contêiner IO para marcar funções impuras que nunca falham.
Essas funções impuras usam data e hora atual, ambiente ou console random :
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_numberAgora podemos ver claramente quais funções são puras e quais são impuras. Isso nos ajuda muito na construção de aplicativos grandes, no teste de unidade do seu código e na composição da lógica de negócios em conjunto.
Como já foi dito, usamos IO quando tratamos de funções que não falham.
E se nossa função falhar e for impura? Como requests.get() que tivemos anteriormente em nosso exemplo.
Então temos que usar um tipo IOResult especial em vez de um Result regular. Vamos encontrar a diferença:
_parse_json sempre retorna o mesmo resultado (espero) para a mesma entrada: você pode analisar json válido ou falhar em um inválido. É por isso que retornamos Result puro, não há IO dentro_make_request é impura e pode falhar. Tente enviar duas solicitações semelhantes com e sem conexão à internet. O resultado será diferente para a mesma entrada. É por isso que devemos usar IOResult aqui: ele pode falhar e tem IOPortanto, para cumprir nosso requisito e separar o código puro do impuro, temos que refatorar nosso exemplo.
Vamos tornar nosso 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 ()E mais tarde podemos usar unsafe_perform_io em algum lugar no nível superior do nosso programa para obter o valor puro (ou "real").
Como resultado desta sessão de refatoração, sabemos tudo sobre nosso código:
Existem vários problemas com código async em Python:
async de uma função sincronizadaawait Os contêineres Future e FutureResult resolvem esses problemas!
A principal característica do Future é que ele permite executar código assíncrono enquanto mantém o contexto de sincronização. Vejamos um exemplo.
Digamos que temos duas funções, a first retorna um número e a second o 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. Se tentarmos apenas executar first() , criaremos apenas uma corrotina inesperada. Não retornará o valor que desejamos.
Mas, se tentarmos executar await first() , precisaremos mudar second para ser async . E às vezes isso não é possível por vários motivos.
No entanto, com Future podemos "fingir" chamar o código assíncrono do código de sincronização:
from returns . future import Future
def second () -> Future [ int ]:
return Future ( first ()). map ( lambda num : num + 1 ) Sem tocar em nossa first função assíncrona ou tornar second função assíncrona, alcançamos nosso objetivo. Agora, nosso valor assíncrono é incrementado dentro de uma função de sincronização.
No entanto, Future ainda precisa ser executado dentro de um eventloop adequado:
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 você pode ver, Future permite trabalhar com funções assíncronas a partir de um contexto de sincronização. E misturar esses dois reinos. Use Future bruto para operações que não podem falhar ou gerar exceções. Praticamente a mesma lógica que tivemos com nosso contêiner IO .
Já abordamos como Result funciona para código puro e impuro. A ideia principal é: não levantamos exceções, nós as retornamos. É especialmente crítico em código assíncrono, porque uma única exceção pode arruinar todas as nossas corrotinas em execução em um único eventloop.
Temos uma combinação útil de contêineres Future e Result : FutureResult . Novamente, isso é exatamente como IOResult , mas para código assíncrono impuro. Use-o quando seu Future puder ter problemas: como solicitações HTTP ou operações de sistema de arquivos.
Você pode facilmente transformar qualquer rotina de lançamento selvagem em um FutureResult calmo:
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 Usar FutureResult manterá seu código protegido contra exceções. Você sempre pode await ou executar dentro de um eventloop qualquer FutureResult para obter a instância de sincronização IOResult para trabalhar com ele de maneira sincronizada.
Anteriormente, você tinha que fazer bastante await ao escrever 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 )Algumas pessoas concordam com isso, mas outras não gostam desse estilo imperativo. O problema é que não houve escolha.
Mas agora você pode fazer a mesma coisa com estilo funcional! Com a ajuda dos contêineres Future e 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 até mesmo algo realmente sofisticado:
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 ),
) Posteriormente também podemos refatorar nossas funções lógicas para serem sincronizadas e retornar FutureResult .
Adorável, não é?
Quer mais? Vá para os documentos! Ou leia estes artigos:
Você tem um artigo para enviar? Sinta-se à vontade para abrir uma solicitação pull!
