让你的函数返回有意义、类型化且安全的内容！
项目描述
让你的函数返回有意义、类型化且安全的内容！
特性
- 将函数式编程引入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 设置 （请参阅此处）.
内容
- Maybe 容器 允许您编写无
None代码 - RequiresContext 容器 允许您使用类型化功能依赖注入
- Result 容器 让您摆脱异常
- IO 容器 和 IOResult 标记所有不纯操作并对它们进行结构化
- Future 容器 和 FutureResult 用于处理
async代码 - 编写您自己的容器！ 您仍将拥有所有适用于您自己的类型的功能（包括现有代码的完全重用和类型安全）
Maybe 容器
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
您可以确保 .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,
)
不是更好吗？
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:情况来捕获所有内容。这样我们最终会捕获到不想要的异常。这种方法可能会长时间隐藏严重的问题。
管道示例
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或者在无效的情况下失败。这就是为什么我们返回纯Result，里面没有IO。 - 我们的
_make_request函数是不纯的并且可能会失败。尝试在有和无网络连接的情况下发送两个相似的请求。结果将会不同。这就是为什么我们必须在这里使用IOResult：它可以失败并且有IO。
因此，为了满足我们的要求并将纯代码与不纯代码分开，我们必须重构我们的例子。
显式IO
让我们使我们的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来获取纯的（或“真实”）值。
通过这次重构，我们了解了关于我们代码的一切
- 哪些部分可能会失败，
- 哪些部分是不纯的，
- 以及如何以智能、可读和类型安全的方式组合它们。
未来容器
Python中的异步代码有几个问题
- 你不能从一个同步函数中调用异步函数
- 任何意外抛出的异常都可能破坏整个事件循环
- 有很多
await语句的丑陋组合
Future和FutureResult容器解决了这些问题！
混合同步和异步代码
Future的主要特点是它允许在保持同步上下文的同时运行异步代码。让我们看看一个例子。
假设我们有两个函数，第一个函数返回一个数字，第二个函数增加它
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改为异步。但有时出于各种原因这是不可能的。
然而，有了Future，我们可以在不触及我们的异步函数first或使second异步的情况下“假装”从同步代码调用异步代码
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实例，并以同步方式使用它。
更好的异步组合
以前，在编写异步代码时，你必须做很多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。
这不是很棒吗？
更多！
想要更多？请访问文档！或者阅读这些文章
你有文章要提交吗？请随意打开一个拉取请求！
下载文件
下载适合您平台的应用程序。如果您不确定该选择哪个，请了解更多关于安装包的信息。
