返回下载 - 返回源码下载

构建状态

让你的函数返回一些有意义的、有类型的、安全的东西！

特征

将函数式编程引入 Python 领域
提供一堆原语来编写声明性业务逻辑
实施更好的架构
完全键入注释并使用mypy检查，兼容 PEP561
添加模拟的更高种类类型支持
提供类型安全的接口来创建您自己的具有强制法律的数据类型
有一堆帮助者可以更好地构图
Pythonic 且写起来和读起来都很愉快？
支持函数和协程，与框架无关
易于入门：有大量文档、测试和教程

立即快速启动！

安装

pip install returns

您还可以使用最新支持的mypy版本安装returns ：

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自由代码
RequiresContext 容器，允许您使用类型化函数依赖注入
结果容器可让您摆脱异常
IO容器和IOResult标记所有不纯的操作并构造它们
Future 容器和 FutureResult 使用async代码
编写你自己的容器！您仍然拥有自己类型的所有功能（包括完整的现有代码重用和类型安全）
使用do-notation使您的代码更容易！

也许是容器

None被称为计算机科学历史上最严重的错误。

那么，我们可以做什么来检查程序中是否存在None ？您可以使用内置的Optional类型并编写很多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

您可以确定不会为Nothing调用.bind_optional()方法。永远忘记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 ,
)

好多了，不是吗？

RequiresContext 容器

许多开发人员确实在 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作为参数！这根本没用！大型代码库将因这一变化而陷入困境。

好的，您可以直接在_award_points_for_letters函数中使用django.settings （或类似的）。并用框架特定的细节破坏你的纯粹逻辑。太丑了！

或者您可以使用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 之外的顶层来捕获几乎所有内容。这样我们最终就会捕获不需要的东西。这种方法可以长期向我们隐藏严重的问题。

管道示例

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 装饰器，我们不再返回常规值，而是返回包装在特殊容器内的值。它将返回 Success[YourType] 或 Failure[Exception]。并且永远不会向我们抛出异常！

我们还使用流和绑定函数来进行方便的声明性组合。

这样我们就可以确保我们的代码不会由于某些隐式异常而在随机位置中断。现在我们控制了所有部分，并为明显的错误做好了准备。

我们还没有完成这个例子，让我们在下一章继续改进它。

IO容器

让我们从另一个角度来看我们的例子。它的所有功能看起来都和正常的一样，乍一看无法分辨它们是纯粹的还是不纯粹的。

这导致了一个非常重要的后果：我们开始将纯代码和不纯代码混合在一起。我们不应该这样做！

当这两个概念混合在一起时，我们在测试或重用它时会遭受非常严重的痛苦。默认情况下，几乎所有内容都应该是纯的。我们应该明确标记程序中不纯净的部分。

这就是为什么我们创建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

正如已经说过的，当我们处理不会失败的函数时，我们会使用IO 。

如果我们的函数可能失败并且不纯净怎么办？就像我们前面的示例中的requests.get()一样。

然后我们必须使用特殊的IOResult类型而不是常规的Result 。我们来找出区别：

对于相同的输入，我们的_parse_json函数总是返回相同的结果（希望如此）：您可以解析有效的json ，也可以解析无效的 json。这就是为什么我们返回纯Result ，里面没有IO
我们的_make_request函数是不纯粹的并且可能会失败。尝试在有互联网连接和没有互联网连接的情况下发送两个类似的请求。对于相同的输入，结果会有所不同。这就是为什么我们必须在这里使用IOResult ：它可能失败并且有IO

因此，为了满足我们的要求并将纯代码与不纯代码分开，我们必须重构我们的示例。

显式IO

让我们明确我们的 IO！

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 中的async代码存在几个问题：

您无法从同步函数调用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 ),
    )