使您的功能返回有意义，打字和安全的东西！
特征
- 将功能性编程带到Python Land
- 提供了许多原始人来编写声明性业务逻辑
- 执行更好的体系结构
- 完全键入注释，并与Mypy兼容
- 添加模拟更高的类型支持
- 提供类型安全的接口，以使用强制定律创建自己的数据类型
- 有一堆助手以获得更好的组成
- Pythonic且愉快地写作和阅读?
- 支持功能和珊瑚酸，框架不可知论
- 容易开始：有很多文档，测试和教程
快速入门！
安装
pip install returns
您还可以使用最新支持的MyPy版本安装returns ：
pip install returns [compatible-mypy]
您还需要配置我们的Mypy插件：
returns _plugin\”>
# In setup.cfg or mypy.ini: [mypy] plugins = returns .contrib.mypy. returns _plugin
或者：
returns _plugin\”]\”>
[ tool . mypy ] plugins = [ \" returns .contrib.mypy. returns _plugin \" ]
我们还建议使用与使用相同的mypy设置，您可以在我们的pyproject.toml文件中的[tool.mypy]部分中找到。
确保您知道如何开始，请查看我们的文档！尝试我们的演示。
内容
- 也许可以让您编写无免费代码的容器
- 需要EnignEsContext容器，该容器允许您使用键入的功能依赖注入
- 结果容器，使您可以摆脱例外
- io容器和路线都标记所有不纯洁的操作并结构它们
- 未来的集装箱和未来可使用异步代码的容器
- 写自己的容器！您仍然拥有自己类型的所有功能（包括完整的现有代码重复使用和类型安全）
- 使用do通知使您的代码更容易！
也许容器
在计算机科学史上，没有一个被称为最严重的错误。
那么，我们该怎么做才能在我们的程序中检查什么？您可以使用内置的可选类型，并写很多，如果不是：条件。但是，在这里和那里进行零检查，使您的代码无法阅读。
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 )
或者您可以使用容器！它由某些类型组成，分别代表现有状态和空状态（而不是无）。
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\”>
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（）方法。忘了永远无关的错误！
我们还可以在容器上绑定可选的返回函数。为此，我们将使用.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 , )
更好，不是吗？
需求context容器
许多开发人员确实在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已深深嵌套。然后，您必须将阈值通过整个呼叫堆栈，包括计算_points和所有其他可能正在途中的功能。他们所有人都必须接受阈值作为参数！这根本没有用！大型代码库将在这种变化中挣扎很多。
好的，您可以在_award_points_for_letters函数中直接使用django.settings（或类似）。并用框架特定的细节破坏您的纯逻辑。那太丑了！
或者您可以使用需求context容器。让我们看看我们的代码如何更改：
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,
)\”>
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_word )( 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 , )
现在，您可以以真正直接和明确的方式传递依赖关系。并具有类型安全，以检查您通过的内容以掩盖背部。查看需要更多信息。在那里，您将学习如何制作\’。也可以配置。
我们还需要对可能失败的与上下文相关的操作进行术语。并且还需要审理，并且需要contextextfutureresult。
结果容器
请确保您也知道面向铁路的编程。
直率的方法
考虑您可以在任何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 ()
似乎合法，不是吗？这似乎也是一个非常简单的代码。您需要的只是模拟请求。要返回所需的结构。
但是，这个微小的代码样本中存在隐藏的问题，几乎看一眼就无法发现。
隐藏的问题
让我们看一下完全相同的代码，但解释了所有隐藏的问题。
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 ()
现在，所有（可能全部？）问题都很明显。我们如何确定此功能可以安全地在我们复杂的业务逻辑中使用？
我们真的不确定！除了捕获预期的例外，我们将不得不创建很多尝试和案例。我们的代码将变得复杂且不可读取！
否则我们可以选择最高级别，除非例外：从字面上捕捉所有内容的情况。这样，我们将最终抓住不需要的人。这种方法可以长期向我们隐藏严重的问题。
管道示例
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()\”>
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 Decorator，我们没有返回定期值，而是返回包裹在特殊容器中的值。它将返回成功[yourtype]或失败[异常]。永远不会给我们施加例外！
我们还将流量和绑定函数用于方便和声明性组成。
这样，由于某些隐式异常，我们可以确定我们的代码不会在随机位置中断。现在，我们控制所有零件，并为明确的错误做好准备。
我们尚未完成此示例，让我们继续在下一章中进行改进。
IO容器
让我们从另一个角度看我们的示例。它的所有功能看起来都像普通的功能：不可能从一见钟觉上判断它们是纯正还是不纯洁。
这导致了一个非常重要的结果：我们开始将纯和不纯净的代码混合在一起。我们不应该那样做！
当这两个概念混合在一起时，我们在测试或重复使用时遭受了很大的痛苦。默认情况下，几乎所有内容都应该是纯净的。而且我们应该明确标记程序中不纯净的部分。
这就是为什么我们创建了IO容器来标记永不失败的不纯粹功能的原因。
这些不纯净的功能使用随机，当前的日期，环境或控制台：
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\”>
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
正如已经说过的那样，当我们处理不会失败的功能时，我们会使用IO。
如果我们的功能会失败并且不纯净怎么办？像requests.get（）一样，我们在示例中较早。
然后，我们必须使用一种特殊的Ioresult类型，而不是常规结果。让我们找到区别：
- 我们的_parse_json函数始终returns相同的结果（希望）相同的输入：您可以解析有效的JSON或失败无效。这就是为什么我们返回纯结果，里面没有IO
- 我们的_make_request函数是不纯净的，可能会失败。尝试在没有Internet连接的情况下发送两个类似的请求。对于相同的输入，结果将有所不同。这就是为什么我们必须在这里使用ioresult：它可能会失败并且有IO
因此，为了满足我们的要求并将纯代码与不纯净的代码分开，我们必须重新分配我们的示例。
显式IO
让我们显式吧！
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()\”>
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来获得纯净（或“真实”）值。
由于这次重构会话，我们了解我们的代码：
- 哪些零件会失败，
- 哪些部分不纯净，
- 如何以智能，可读和类型的方式组成它们。
未来的容器
Python中的异步代码有几个问题：
- 您不能从同步一个人调用异步函数
- 任何出乎意料的抛出例外都可能破坏您的整个活动循环
- 丑陋的构图，有很多等待的陈述
未来和未来的容器解决了这些问题！
混合同步和异步代码
未来的主要特征是，它允许在维护同步上下文的同时运行异步代码。让我们看看一个例子。
假设我们有两个功能，第一个函数returns一个数字，第二个功能将其递增：
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.
如果我们尝试首先运行（），我们将创建一个未知的Coroutine。它不会返回我们想要的价值。
但是，如果我们尝试首先运行（），那么我们需要更改第二个以使其成为异步。有时由于各种原因不可能。
但是，随着未来，我们可以“假装”从同步代码调用异步代码：
returns.future import Future
def second() -> Future[int]:
return Future(first()).map(lambda num: num + 1)\”>
from returns . future import Future def second () -> Future [ int ]: return Future ( first ()). map ( lambda num : num + 1 )
没有触及我们的第一个异步功能或制作第二异步，我们就实现了我们的目标。现在，我们的异步值在同步函数中增加了。
但是，未来仍然需要在适当的Eventloop中执行：
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
如您所见，未来可以从同步上下文中使用异步函数。并将这两个领域混合在一起。使用原始未来进行无法失败或提出异常的操作。与IO容器相同的逻辑几乎相同。
异步代码没有例外
我们已经介绍了结果如何用于纯净和不纯正的代码。主要想法是：我们不提出例外，我们退还它们。这对于异步代码尤为重要，因为一个例外可能会破坏我们在单个Eventloop中运行的所有Coroutines。
我们有一个方便的未来和结果容器的组合：未来。同样，这与ioresult完全一样，但对于不纯净的异步代码。当您的未来可能会遇到问题时使用它：例如HTTP请求或文件系统操作。
您可以轻松地将任何狂野的抛出式冠军变成一个平静的未来：
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\”>
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
使用FutureSult将使您的代码免受例外情况。您始终可以在Eventloop中等待或执行任何未来的未来，以获得同步实例以同步方式使用它。
更好的异步组成
以前，您在编写异步代码时必须进行很多等待：
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 )
有些人可以接受，但是有些人不喜欢这种命令风格。问题是别无选择。
但是现在，您可以在功能风格上做同样的事情！在未来和未来的容器的帮助下：
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(…),
)\”>
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 (...), )
甚至是真正花哨的东西：
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),
)\”>
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 ), )
稍后，我们还可以重构我们的逻辑功能是同步并返回未来的功能。
可爱，不是吗？
更多的！
想要更多吗？去文档！或阅读这些文章：
- python例外认为是抗模式
- 在Python中执行单一责任原则
- python中的键入功能依赖注入
- 异步应该是多么的
- Python中更高的类型
- 使测试成为应用程序的一部分
您有一篇文章要提交吗？随时打开拉动请求！
