returns
Version 0.24.0
Пусть ваши функции возвращают что-то осмысленное, типизированное и безопасное!
mypy , совместим с PEP561.Быстрый старт прямо сейчас!
pip install returns Вы также можете установить returns с последней поддерживаемой версией mypy :
pip install returns[compatible-mypy] Вам также потребуется настроить наш плагин mypy :
# In setup.cfg or mypy.ini:
[mypy]
plugins =
returns.contrib.mypy.returns_pluginили:
[ tool . mypy ]
plugins = [ " returns.contrib.mypy.returns_plugin " ] Мы также рекомендуем использовать те же настройки mypy , что и мы.
Убедитесь, что вы знаете, с чего начать, ознакомьтесь с нашей документацией! Попробуйте нашу демо-версию.
None -свободный код.async кодом.do-notation чтобы упростить код! None нельзя назвать худшей ошибкой в истории информатики.
Итак, что мы можем сделать, чтобы проверить наличие None в наших программах? Вы можете использовать встроенный тип «Необязательный» и писать множество условий if some is not None: ». Но наличие null проверок здесь и там делает ваш код нечитаемым .
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 ) Или вы можете использовать контейнер Maybe! Он состоит из типов Some и Nothing , представляющих существующее состояние и пустое (вместо 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 Вы можете быть уверены, что метод .bind_optional() не будет вызываться Nothing . Забудьте об ошибках, связанных с None навсегда!
Мы также можем привязать Optional возвращающую функцию к контейнеру. Для этого мы собираемся использовать метод .bind_optional .
А вот как будет выглядеть ваш первоначальный рефакторинг кода:
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 ,
)Гораздо лучше, не так ли?
Многие разработчики используют в Python своего рода внедрение зависимостей. И обычно это основано на идее, что существует некий контейнер и процесс сборки.
Функциональный подход намного проще!
Представьте, что у вас есть игра на основе django , в которой вы начисляете пользователям баллы за каждую угаданную букву в слове (неугаданные буквы помечаются знаком '.' ):
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!Потрясающий! Это работает, пользователи довольны, ваша логика чиста и великолепна. Но позже вы решите сделать игру более увлекательной: давайте сделаем минимальный порог ответственных писем настраиваемым для дополнительной задачи.
Вы можете просто сделать это напрямую:
def _award_points_for_letters ( guessed : int , threshold : int ) -> int :
return 0 if guessed < threshold else guessed Проблема в том, что _award_points_for_letters глубоко вложены. А затем вам придется передать threshold через весь стек вызовов, включая calculate_points и все другие функции, которые могут быть в пути. Всем им придется принять threshold в качестве параметра! Это совершенно не полезно! Большие базы кода будут испытывать большие трудности из-за этого изменения.
Хорошо, вы можете напрямую использовать django.settings (или аналогичный) в своей функции _award_points_for_letters . И разрушьте свою чистую логику деталями, специфичными для фреймворка . Это некрасиво!
Или вы можете использовать контейнер RequiresContext . Посмотрим, как изменится наш код:
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 ,
) И теперь вы можете передавать свои зависимости действительно напрямую и явно. И имейте типобезопасность, чтобы проверять то, что вы передаете, чтобы прикрыть свою спину. Дополнительную информацию можно найти в документации RequiresContext. Там вы научитесь делать '.' также настраивается.
У нас также есть RequiresContextResult для операций, связанных с контекстом, которые могут завершиться неудачей. А также RequiresContextIOResult и RequiresContextFutureResult.
Пожалуйста, убедитесь, что вы также знакомы с программированием, ориентированным на железные дороги.
Рассмотрим этот код, который вы можете найти в любом проекте 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 () Кажется законным, не так ли? Это также кажется довольно простым кодом для тестирования. Все, что вам нужно, это издеваться над requests.get , чтобы вернуть нужную вам структуру.
Но в этом крошечном примере кода есть скрытые проблемы, которые практически невозможно обнаружить с первого взгляда.
Давайте посмотрим на тот же код, но с объяснением всех скрытых проблем.
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 ()Теперь все (вероятно, все?) проблемы ясны. Как мы можем быть уверены, что эту функцию можно будет безопасно использовать внутри нашей сложной бизнес-логики?
Мы действительно не можем быть уверены! Нам придется создать множество try и except , чтобы перехватить ожидаемые исключения. Из-за всего этого бардака наш код станет сложным и нечитабельным!
Или мы можем использовать верхний уровень, except Exception: case, чтобы перехватывать буквально все. И таким образом мы в конечном итоге поймаем нежелательных. Такой подход может надолго скрыть от нас серьезные проблемы.
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 ()Теперь у нас есть чистый, безопасный и декларативный способ выразить потребности нашего бизнеса:
Теперь вместо возврата обычных значений мы возвращаем значения, заключенные в специальный контейнер благодаря декоратору @safe. Он вернет Success[YourType] или Fail[Exception]. И никогда не выкинет нам исключение!
Мы также используем функции потока и привязки для удобной и декларативной композиции.
Таким образом, мы можем быть уверены, что наш код не сломается в случайных местах из-за какого-либо неявного исключения. Теперь мы контролируем все детали и готовы к явным ошибкам.
Мы еще не закончили с этим примером, давайте продолжим его улучшать в следующей главе.
Давайте посмотрим на наш пример под другим углом. Все его функции выглядят как обычные: с первого взгляда невозможно определить, чистые они или нечистые.
Это приводит к очень важному последствию: мы начинаем смешивать чистый и нечистый код вместе . Мы не должны этого делать!
Когда эти две концепции смешиваются, мы очень сильно страдаем при тестировании или повторном использовании. По умолчанию почти все должно быть чистым. И мы должны явно отмечать нечистые части программы.
Вот почему мы создали контейнер IO , чтобы отмечать нечистые функции, которые никогда не выходят из строя.
Эти нечистые функции используют 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_numberТеперь мы можем ясно видеть, какие функции чистые, а какие нечистые. Это очень помогает нам в создании больших приложений, модульном тестировании вашего кода и совместном составлении бизнес-логики.
Как уже было сказано, мы используем IO , когда обрабатываем функции, которые не дают сбоя.
Что, если наша функция может дать сбой и окажется нечистой? Как и в случае с requests.get() который мы использовали ранее в нашем примере.
Тогда нам придется использовать специальный тип IOResult вместо обычного Result . Найдем разницу:
_parse_json всегда возвращает один и тот же результат (надеюсь) для одного и того же ввода: вы можете либо проанализировать действительный json , либо потерпеть неудачу при недопустимом. Вот почему мы возвращаем чистый Result , внутри нет IO_make_request нечиста и может дать сбой. Попробуйте отправить два одинаковых запроса с подключением к Интернету и без него. Результат будет разным для одного и того же ввода. Вот почему мы должны использовать здесь IOResult : он может дать сбой и имеет IOИтак, чтобы выполнить наше требование и отделить чистый код от нечистого, нам придется провести рефакторинг нашего примера.
Давайте сделаем наш ввод-вывод явным!
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 ()А позже мы можем использовать unsafe_perform_io где-нибудь на верхнем уровне нашей программы, чтобы получить чистое (или «настоящее») значение.
В результате этого сеанса рефакторинга мы знаем все о нашем коде:
Есть несколько проблем с async кодом в Python:
async функцию из синхронизирующей.await Контейнеры Future и FutureResult решают эти проблемы!
Основная особенность Future заключается в том, что он позволяет запускать асинхронный код, сохраняя контекст синхронизации. Давайте посмотрим пример.
Допустим, у нас есть две функции: first возвращает число, а second увеличивает его:
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. Если мы попытаемся просто запустить first() , мы просто создадим неожиданную сопрограмму. Он не вернет нужное нам значение.
Но если мы попытаемся запустить await first() , нам нужно будет изменить second на async . А иногда это невозможно по разным причинам.
Однако с помощью Future мы можем «притвориться», что вызываем асинхронный код из кода синхронизации:
from returns . future import Future
def second () -> Future [ int ]:
return Future ( first ()). map ( lambda num : num + 1 ) Не трогая нашу first асинхронную функцию и не создавая second асинхронную функцию, мы достигли своей цели. Теперь наше асинхронное значение увеличивается внутри функции синхронизации.
Однако Future по-прежнему требует выполнения внутри правильного цикла событий:
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 Как видите, Future позволяет работать с асинхронными функциями из контекста синхронизации. И смешать эти два мира вместе. Используйте необработанное Future для операций, которые не могут завершиться сбоем или вызвать исключения. Практически та же логика, что и с нашим контейнером IO .
Мы уже рассмотрели, как Result работает как для чистого, так и для нечистого кода. Основная идея такова: мы не вызываем исключения, мы их возвращаем. Это особенно важно в асинхронном коде, поскольку одно исключение может разрушить все наши сопрограммы, работающие в одном цикле событий.
У нас есть удобная комбинация контейнеров Future и Result : FutureResult . Опять же, это точно так же, как IOResult , но для нечистого асинхронного кода. Используйте его, когда у вашего Future могут возникнуть проблемы: например, HTTP-запросы или операции с файловой системой.
Вы можете легко превратить любую дикую сопрограмму, вызывающую бросок, в спокойную FutureResult :
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 Использование FutureResult защитит ваш код от исключений. Вы всегда можете await или выполнить внутри цикла событий любой FutureResult , чтобы синхронизировать экземпляр IOResult для синхронной работы с ним.
Раньше при написании async кода приходилось довольно долго 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 )Некоторых это устраивает, но некоторым не нравится этот императивный стиль. Проблема в том, что выбора не было.
Но теперь вы можете сделать то же самое в функциональном стиле! С помощью контейнеров Future и 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 (...),
)Или даже что-то очень необычное:
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 ),
) Позже мы также сможем провести рефакторинг наших логических функций для синхронизации и возврата FutureResult .
Прекрасно, не так ли?
Хотите больше? Идите в документы! Или прочитайте эти статьи:
У вас есть статья для отправки? Смело открывайте запрос на вытягивание!
