Python中的函数式编程

您可能还希望正确配置mypy并将我们的插件安装到修复此现有问题：

那么，我们可以做些什么来检查我们的程序中是否没有空值呢？您可以使用内置的可选类型，并编写很多条件(如果有些不是None的话)：条件。但是，在这里和那里进行NULL检查会使您的代码不可读。

USER：可选[USER]如果USER不是无：BALANCE=USER。如果BALANCE不是NONE，则GET_BALANCE()：BALANCE_Credit=B。如果Balance_Credit不是NONE且Balance_Credit>；0：Can_Buy_Stuff=True，则Credit_Amount()：Can_Buy_Stuff=False：Can_Buy_Stuff=False。

或者您可以使用Maybe CONTAINER！它由一些和Nothing类型组成，分别表示现有状态和空(而不是None)状态。

从返回键入IMPORTIONAL OPTIONAL。可能导入可能，可能@可能#修饰符将现有可选的[int]转换为可能的[int]def BAD_Function()->；可选的[int]：.。可能_结果：可能[浮点]=BAD_Function()。MAP(lambda number：number/2，)#=>；可能仅当存在非None值时才返回某个[Float]，否则将不返回任何内容。

您可以确定.map()方法不会因为什么都不会被调用。永远忘掉与此无关的错误吧！

USER：可选[USER]CAN_BUY_STUSITH：可能[bool]=可能。FROM_VALUE(用户)。map(#type hint不是必需的，lambda REAL_USER：REAL_USER。get_Balance()，)。MAP(λ平衡：平衡。Credit_Amount()，)。map(lambda Balance_Credit：Balance_Credit>；0，)。

许多开发人员确实在Python中使用了某种依赖注入，而且通常是基于这样一种想法，即存在某种容器和组装过程。

假设您有一个基于Django的游戏，您为用户在一个单词中的每个猜到的字母奖励积分(未猜到的字母标记为'；.'；)：

从姜戈来的。HTTP从WORD_APP导入HttpRequest、HttpResponse。逻辑导入Calculate_Points def view(Request：HttpRequest)->；HttpResponse：USER_WORD：STR=REQUEST。POST['；Word'；]#Just a Example Points=Calculate_Points(User_Word).#之后，您可以在`WORD_APP/logic.py`：def Calculate_Points(Word：Str)->；int：[Letter for Letter if Letter！='；.'；中的字母！='；.'；])RETURN_INDUTE_POINTS_FOR_LETERS(GUIESSED_LETERS_COUNT)def_INDUTE_POINTS_FOR_LETERS(猜想：int)->；int：如果猜到，则返回0<；5，否则猜#至少6分！

太棒了！它很管用，用户很高兴，你的逻辑是纯粹而令人敬畏的。但是，后来你决定让这个游戏更有趣：让我们使最小的可识别字母阈值可配置用于额外的挑战。

DEF_INDUTE_POINTS_FOR_LETERS(猜测：INT，THRESHOLD：INT)->；INT：如果猜测<；THRESHOLD则返回0。

问题是_rediate_points_for_letters是深度嵌套的。然后，您必须通过整个调用堆栈传递阈值，包括Calculate_Points和所有其他可能正在进行的函数。所有这些函数都必须接受阈值作为参数！这根本没有用！大型代码库将很难从这个更改中获益。

好的，您可以直接在您的_INDUTE_POINTS_FOR_LETERS函数中使用django.settings(或类似的)，然后用框架特定的细节来破坏您的纯逻辑。那太难看了！

从姜戈来的。Django中的会议导入设置。HTTP从WORD_APP导入HttpRequest、HttpResponse。逻辑导入Calculate_Points def view(Request：HttpRequest)->；HttpResponse：USER_WORD：STR=REQUEST。post['；word'；]#Just a Example Points=Calculate_Points(USER_WORD)(SETTINGS)#传递依赖关系.#之后，在您的`WORD_APP/logic.py`：from Typing_Extensions import Protocol from Returns中的某个地方，以某种方式向用户显示结果。Context import RequiresContext CLASS_Dep(协议)：#我们依赖抽象，而不是直接的值或类型WORD_THRESHOLD：int def Calculate_Points(Word：Str)->；RequiresContext[_Dep，int]：GUESSED_LETERS_COUNT=len([Letter for Letter if Letter！='；.'；])RETURN_INDUTE_POINTS_FOR_Letters(GUSED_Letters_Count)def_INDUTE_POINTS_FOR_Letters。RequiresContext[_Dep，int]：返回RequiresContext(如果猜测<；deps，则lambda deps：0。WORD_THRESHOLD ELSE猜测，)。

现在，您可以真正直接而显式地传递依赖项，并且具有类型安全功能来检查您传递的内容以掩护您的后背。请查看RequiresContext文档以了解更多信息。在那里，您将学习如何使'；.'；也是可配置的。

导入请求def FETCH_USER_PROFILE(user_id：int)-&>；'；UserProfile'；："；"；"；从外部API获取UserProfile判决。"；"；"；Response=Requests。GET('；/api/Users/{0}'；。format(User_Id))响应。RAISE_FOR_STATUS()返回响应。json()。

看起来是合法的，不是吗？它看起来也像是要测试的非常简单的代码。所有您需要做的就是模拟请求。返回您需要的结构。

但是，在这个微小的代码样本中有隐藏的问题，乍一看几乎不可能发现这些问题。

让我们来看看完全相同的代码，但要解释所有隐藏的问题。

导入请求def FETCH_USER_PROFILE(user_id：int)-&>；'；UserProfile'；："；"；"；从外部API获取UserProfile判决。"；"；"；Response=Requests。GET('；/api/Users/{0}'；。format(User_Id))#如果我们尝试查找不存在的用户怎么办？#否则网络将关闭？否则服务器将返回500？#，在这种情况下，下一行将失败并出现异常。#我们需要处理此函数中可能出现的所有错误#，不要向消费者返回损坏的数据。回应。raise_for_status()#如果我们收到无效的JSON怎么办？#下一行将引发异常！返回响应。json()。

现在，全部(可能全部？)。问题是显而易见的，我们如何确保在复杂的业务逻辑中使用此函数是安全的呢？

我们真的不能确定！我们将不得不创建大量的尝试和排除案例-仅仅是为了捕捉预期的异常。

从退货导入请求。结果导入结果，安全不退货。来自退货的管道导入流。无指针导入绑定def fetch_user_profile(user_id：int)->；result['；UserProfile'；，Exception]："；"；"；从外部API获取`UserProfile`TypedDict。"；"；"；返回流(user_id，_make_request，bind(_Parse_Json)，)@safe def_make_request(user_id。Response：#TODO：这个示例还没有结束，请阅读更多关于`IO`：Response=Requests的内容。GET('；/api/Users/{0}'；。format(User_Id))响应。raise_for_status()return response@safe def_parse_json(Response：Requests。响应)->；'；UserProfile'；：返回响应。json()

现在，我们有了一种干净、安全和声明的方式来表达我们的业务需求：

现在，多亏了@safedecorator，我们返回了包装在特殊容器中的值，而不是返回常规值。它将返回Success[YourType]或Failure[Exception]，并且永远不会向我们抛出异常！

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

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

让我们从另一个角度来看我们的例子。它的所有功能看起来都很普通：乍一看是纯的还是不纯的。

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

当这两个概念混合在一起时，我们在测试或重用它时会受到很大的影响，几乎所有的东西在默认情况下都应该是纯的，而且我们应该明确地标记程序中不纯的部分。

将随机导入日期时间作为DT从返回导入。IO import IO def get_Random_Number()->；IO[int]：#或使用`@impure`修饰符返回IO(Random.。randint(1，10))#is'；不是纯的，因为现在随机：Callable[[]，IO[dt.datetime]]=unure(dt.datetime.。NOW)@unure def return_and_show_next_number(previous：int)->；int：next_number=previous+1 print(Next_Number)#不是纯的，因为IO返回NEXT_NUMBER

现在我们可以清楚地看到哪些函数是纯函数，哪些函数是错误函数，这对我们构建大型应用程序、对代码进行单元测试以及共同组成业务逻辑有很大帮助。

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

如果我们的函数可能失败并且不纯呢？就像我们在前面的示例中使用的requests.get()一样。

对于相同的输入，我们的_parse_json函数总是返回相同的结果(希望如此)：您可以解析有效的json，也可以解析无效的json失败。这就是我们返回纯结果的原因。

我们的_make_request函数不纯，可能会失败。尝试在连接和不连接Internet的情况下发送两个类似的请求。对于相同的输入，结果会有所不同。这就是我们必须在此处使用IOResult的原因。

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

从退货导入请求。IO从返回导入IO、IOResult、INPURE_SAFE。结果导入安全，不会退货。来自退货的管道导入流。无指针导入BIND_RESULT def FETCH_USER_PROFILE(user_id：int)->；IOResult['；UserProfile'；，Exception]："；"；"；从外部API检索`UserProfile`TypedDict。"；"；"；Return Flow(user_id，_make_request，#BEFORE：def(Response)->；UserProfile#After。ResultE[UserProfile]#After Bind_ResultE：def(IOResultE[Response])->；IOResultE[UserProfile]bind_result(_Parse_Json)，)@unure_safe def_make_request(user_id：int)->；请求。响应：响应=请求。GET('；/api/Users/{0}'；。format(User_Id))响应。raise_for_status()return response@safe def_parse_json(Response：Requests。响应)->；'；UserProfile'；：返回响应。json()。

然后，我们可以在程序的顶层取消Safe_Performance_iosomewhere，以获得纯值。

未来的主要特点是，它允许在保持同步上下文的同时运行异步代码。让我们来看一个例子。

假设我们有两个函数，第一个函数返回一个数字，第二个函数使其递增：

异步def first()->；int：返回1 def Second()：#从这里怎么调用`first()`？返回FIRST()+1#BOOM！不要这样做，这是错误的。这只是个例子。

如果我们尝试只运行first()，我们只会创建一个意外的协议表，它不会返回我们想要的值。

但是，如果我们尝试先运行await first()，那么我们需要将Second更改为异步。有时，由于各种原因，这是不可能的。

没有接触我们的第一个异步函数或使第二个异步函数，我们已经实现了我们的目标。现在，我们的异步值在同步函数中递增。

import anyio#或asyncio或任何其他库#然后我们可以将我们的‘Future’传递到任何库：asyncio、trio、curio。#并使用任何事件循环：Regular、uvloop，甚至是自定义循环，等等，断言anyio。运行(Second()。可等待)==2。

如您所见，Future允许您在同步上下文中使用异步函数。并将这两个领域混合在一起。对不能引发异常的操作使用RAW Future。

我们已经讨论了结果是如何工作的，其主要思想是：我们不引发异常，而是返回它们。这在异步代码中尤其关键，因为单个异常可能会破坏在单个事件循环中运行的所有协程。

我们有一个很方便的Future和Result容器组合：FutureResult.当您的Future可能有问题时使用它：比如HTTP请求或文件系统操作。

从退货导入Anio。将来从退货导入FURRENT_SAFE。IO从退货导入IOFailure。管道导入IS_SUCCESS@FORUD_SAFE异步定义提升()：提升ValueError('；别急！'；)ioresult=anyio。奔跑(举起。可等待)#all`Future的返回IO容器Assert NOT_SUCCESS(Ioresult)#True Assert ioresult==IOFailure(ValueError('；别这么快！'；))#Also True。

使用FutureResult将保护您的代码不会出现异常。您可以始终在EventLoop中等待或执行任何FutureResult以获取同步IOResult实例以同步方式使用它。

异步def FETCH_USER(USER_ID：INT)->；'；USER'；：.。异步定义GET_USER_PERMISSIONS(USER：'；USER'；)-&>&39；权限'；：.。异步定义SECURE_ALLOWED(权限：'；权限'；)->；bool：.。Async def main(user_id：int)->；bool：#另外，不要忘记使用`try/except`处理所有可能的错误！user=await FETCH_USER(User_Id)#我们每次使用CORO时都会等待！权限=等待GET_USER_PERMISSIONS(用户)返回等待确保_允许(权限)。

有些人可以接受，但有些人不喜欢这种命令式的风格。问题是别无选择。

但是现在，借助Future和FutureResult容器，您可以在函数风格中做同样的事情！

从退货导入Anio。将来导入FutureResultE，FORMURE_SAFE FORMERS FORMENT。IO导入IOSuccess，IOFailure@Future_Safe异步DEF FETCH_USER(user_id：int)->；'；USER'；：.@Future_Safe异步DEF GET_USER_PERSIONS(USER：'；USER'；)-&>&39；权限'；：.@FORFORY_SAFE ASYNC DEF SUBURE_ALLOWED(权限：'；权限'；)->。def main(user_id：int)->；FutureResultE[bool]：#现在可以将`main`变成同步函数，它根本不会`wait`。#我们也不再关心异常，它们已经被处理了。返回FETCH_USER(USER_ID)。绑定(GET_USER_PERMISSIONS)。绑定(确保_允许)正确的用户标识：int#具有所需的权限禁用用户标识：int#没有所需的权限错误的用户标识：int#不存在#我们可以有正确的业务结果：断言anyio。运行(main(Right_User_Id)。可等待)==IOSuccess(True)断言anyio。运行(Main(BIREBLED_USER_ID)。waitable)==IOSuccess(False)#否则我们可能会遇到错误：断言anyio。运行(main(Error_User_Id)。可等待)==IOFailure(UserDoesNotExistError(.)，)

从退货中。从返回的无指针导入绑定。管道导入流程def main(user_id：int)->；FutureResultE[bool]：Return Flow(FETCH_USER(User_Id)，BIND(GET_USER_PERMISSIONS)，BIND(Assure_Allowed)，)