Python应该有什么异步

在过去的几年中，async异步编程关键字和语义已经渗透到许多流行的编程语言：JavaScript的，铁锈，C＃ ，和其他许多人。当然，Python也有它async/await，它是在Python 3.5中引入的。



在本文中，我想讨论异步代码的问题，推测替代方案，并提出一种同时支持同步和异步应用程序的新方法。

功能颜色

当编程语言中包含异步函数时，它实际上分为两部分。出现红色功能（或异步），而某些功能保持蓝色（同步）。



主要问题是蓝色函数不能调用红色，但是红色可能导致蓝色。例如，在Python中，这部分是正确的：异步函数只能调用同步非阻塞函数。但是无法从描述中确定该功能是否正在阻塞。 Python是一种脚本语言。



这种分裂导致语言分为两个子集：同步和异步。 Python 3.5是五年前发布的，但是async它仍然不如Python的同步功能那样得到很好的支持。



您可以在这篇很棒的文章中阅读更多有关功能颜色的信息。

重复的代码

实际上，不同颜色的功能意味着代码重复。



假设您正在开发一个用于检索网页大小的CLI工具，并且您想要同时维护同步和异步方式。例如，如果您正在编写库并且不知道如何使用代码，则这是必需的。这不仅涉及PyPI库，还涉及我们自己的库，这些库具有用于各种服务的通用逻辑，例如用Django和aiohttp编写。尽管，当然，独立的应用程序大多只以同步方式或异步方式编写。



让我们从同步伪代码开始：

def fetch_resource_size(url: str) -> int:
    response = client_get(url)
    return len(response.content)

看起来不错。现在让我们看一下异步模拟：

async def fetch_resource_size(url: str) -> int:
    response = await client_get(url)
    return len(response.content)

通常，这是相同的代码，但增加了单词async和await。而且我没有补全-比较httpx教程中的代码示例：

• 同步码;
• 异步代码。

有完全相同的图片。

抽象和组成

事实证明，您需要重写所有同步代码并在此处和此处进行排列async，await以使程序成为异步的。



两条原则可以帮助解决此问题。首先，让我们将命令式伪代码重写为函数式。这样可以使您更清楚地看到图片。

def fetch_resource_size(url: str) -> Abstraction[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

您问这种方法是.map什么，它有什么作用。这就是复杂抽象和纯函数的组合如何以函数样式出现的方式。这使您可以从现有状态创建具有新状态的新抽象。假设它client_get(url)最初返回Abstraction[Response]，并且调用.map(lambda response: len(response.content))将响应转换为所需的实例Abstraction[int]。



清楚下一步该做什么。请注意，从几个独立的步骤转移到顺序函数调用是多么容易。此外，我们更改了响应的类型：现在，该函数返回了一些抽象。



让我们重写代码以使用异步版本：

def fetch_resource_size(url: str) -> AsyncAbstraction[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

唯一不同的是返回类型- AsyncAbstraction。其余代码完全相同。您不再需要使用关键字async和await。await根本不使用（因为一切都已开始），并且没有它就没有意义async。



最后一件事是确定我们需要哪个客户端：异步还是同步。

def fetch_resource_size(
    client_get: Callable[[str], AbstactionType[Response]],
    url: str,
) -> AbstactionType[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

client_get现在是可调用的类型参数，该参数将URL字符串作为输入并AbstractionType在对象上返回某种类型Response。AbstractionType-Abstraction或AsyncAbstraction来自先前示例。



当我们通过时Abstraction，代码将同步运行，当AsyncAbstraction-相同的代码自动开始异步运行。

IOResult和FutureResult

幸运的是，dry-python/returns正确的抽象已经就位。



让我向您介绍一种类型安全，对mypy友好，与框架无关的完全Python工具。它具有令人敬畏的，舒适的，奇妙的抽象，几乎可以在任何项目中使用。

同步选项

首先，我们将添加依赖项以获得可重现的示例。

pip install returns httpx anyio

接下来，让我们将伪代码转换为有效的Python代码。让我们从sync选项开始。

from typing import Callable
 
import httpx
 
from returns.io import IOResultE, impure_safe
 
def fetch_resource_size(
    client_get: Callable[[str], IOResultE[httpx.Response]],
    url: str,
) -> IOResultE[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )
 
print(fetch_resource_size(
    impure_safe(httpx.get),
    'https://sobolevn.me',
))
# => <IOResult: <Success: 27972>>

进行了一些更改才能获得有效的代码：

• 使用IOResultE是处理同步IO错误（异常并非总是有效）的一种功能性方法。基于的类型Result允许您模拟异常，但要使用单独的值Failure()。然后将成功的退出包装为type Success。通常没有人关心异常，但我们会这样做。
• 使用httpx它可以处理同步和异步请求。
• 使用函数impure_safe将返回类型httpx.get转换为抽象IOResultE。

异步选项

让我们尝试在异步代码中执行相同的操作。

from typing import Callable
 
import anyio
import httpx
 
from returns.future import FutureResultE, future_safe
 
def fetch_resource_size(
    client_get: Callable[[str], FutureResultE[httpx.Response]],
    url: str,
) -> FutureResultE[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )
 
page_size = fetch_resource_size(
    future_safe(httpx.AsyncClient().get),
    'https://sobolevn.me',
)
print(page_size)
print(anyio.run(page_size.awaitable))
# => <FutureResult: <coroutine object async_map at 0x10b17c320>>
# => <IOResult: <Success: 27972>>

您会看到：结果完全相同，但是现在代码正在异步运行。但是，其主要部分没有改变。但是，您需要注意以下几点：

• 同时IOResultE改为异步FutureResultE，impure_safe-上future_safe。它的工作原理相同，但是返回不同的抽象：FutureResultE。
• 使用AsyncClient从httpx。
• FutureResult由于红色函数无法自行调用，因此需要运行结果值。
• 该实用程序anyio是用来表明该方法适用于任何异步库：asyncio，trio，curio。

二合一

我将向您展示如何在一个类型安全的API中结合同步和异步版本。尚未发布用于IO的



高级类型和类型类（它们将出现在0.15.0中），因此我将在通常的示例中进行说明@overload：

from typing import Callable, Union, overload
 
import anyio
import httpx
 
from returns.future import FutureResultE, future_safe
from returns.io import IOResultE, impure_safe
 
@overload
def fetch_resource_size(
    client_get: Callable[[str], IOResultE[httpx.Response]],
    url: str,
) -> IOResultE[int]:
    """Sync case."""
 
@overload
def fetch_resource_size(
    client_get: Callable[[str], FutureResultE[httpx.Response]],
    url: str,
) -> FutureResultE[int]:
    """Async case."""
 
def fetch_resource_size(
    client_get: Union[
        Callable[[str], IOResultE[httpx.Response]],
        Callable[[str], FutureResultE[httpx.Response]],
    ],
    url: str,
) -> Union[IOResultE[int], FutureResultE[int]]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

我们使用修饰符来@overload描述允许输入的数据以及返回值的类型。您@overload可以在另一篇文章中了解有关装饰器的更多信息。



使用同步或异步客户端的函数调用如下所示：

# Sync:
print(fetch_resource_size(
    impure_safe(httpx.get),
    'https://sobolevn.me',
))
# => <IOResult: <Success: 27972>>
 
# Async:
page_size = fetch_resource_size(
    future_safe(httpx.AsyncClient().get),
    'https://sobolevn.me',
)
print(page_size)
print(anyio.run(page_size.awaitable))
# => <FutureResult: <coroutine object async_map at 0x10b17c320>>
# => <IOResult: <Success: 27972>>

如您所见，fetch_resource_size在同步版本中，它会立即返回IOResult并执行它。在异步版本中，像常规协程一样，需要一个事件循环。anyio用于显示结果。



在mypy此代码中没有注释：

» mypy async_and_sync.py
Success: no issues found in 1 source file

让我们看看如果发生了什么事情会发生什么。

---lambda response: len(response.content),
+++lambda response: response.content,

mypy 轻松发现新错误：

» mypy async_and_sync.py
async_and_sync.py:33: error: Argument 1 to "map" of "IOResult" has incompatible type "Callable[[Response], bytes]"; expected "Callable[[Response], int]"
async_and_sync.py:33: error: Argument 1 to "map" of "FutureResult" has incompatible type "Callable[[Response], bytes]"; expected "Callable[[Response], int]"
async_and_sync.py:33: error: Incompatible return value type (got "bytes", expected "int")

灵巧的手，没有魔术：用正确的抽象编写异步代码只需要简单的旧结构即可。但是，我们为不同类型获得相同的API的事实确实很棒。例如，它允许您从HTTP请求的工作方式中抽象：同步还是异步。



希望该示例显示了异步程序的真正表现。如果尝试使用dry-python / return，则会发现更多有趣的事情。在新版本中，我们已经为使用高级类型类型和所有必要的接口制作了必要的原语。上面的代码现在可以像这样重写：

from typing import Callable, TypeVar

import anyio
import httpx

from returns.future import future_safe
from returns.interfaces.specific.ioresult import IOResultLike2
from returns.io import impure_safe
from returns.primitives.hkt import Kind2, kinded

_IOKind = TypeVar('_IOKind', bound=IOResultLike2)

@kinded
def fetch_resource_size(
    client_get: Callable[[str], Kind2[_IOKind, httpx.Response, Exception]],
    url: str,
) -> Kind2[_IOKind, int, Exception]:
    return client_get(url).map(
        lambda response: len(response.content),
    )


# Sync:
print(fetch_resource_size(
    impure_safe(httpx.get),
    'https://sobolevn.me',
))
# => <IOResult: <Success: 27972>>

# Async:
page_size = fetch_resource_size(
    future_safe(httpx.AsyncClient().get),
    'https://sobolevn.me',
)
print(page_size)
print(anyio.run(page_size.awaitable))
# => <FutureResult: <coroutine object async_map at 0x10b17c320>>
# => <IOResult: <Success: 27972>>

查看`master`分支，它已经在那儿工作了。

更多干蟒蛇功能

这是我最引以为豪的其他一些有用的干Python功能。

• 键入函数partial和@curry。

from returns.curry import curry, partial
 
def example(a: int, b: str) -> float:
    ...
 
reveal_type(partial(example, 1))
# note: Revealed type is 'def (b: builtins.str) -> builtins.float'
 
reveal_type(curry(example))
# note: Revealed type is 'Overload(def (a: builtins.int) -> def (b: builtins.str) -> builtins.float, def (a: builtins.int, b: builtins.str) -> builtins.float)'

@curry例如， 这使您可以使用，例如：

@curry
def example(a: int, b: str) -> float:
    return float(a + len(b))
 
assert example(1, 'abc') == 4.0
assert example(1)('abc') == 4.0

• 具有类型推断的功能管道。

使用自定义的mypy插件，您可以构建返回类型的功能管道。

from returns.pipeline import flow
assert flow(
    [1, 2, 3],
    lambda collection: max(collection),
    lambda max_number: -max_number,
) == -3

通常在类型化代码中，使用lambda很不方便，因为它们的参数始终是type Any。推论mypy解决了这个问题。



有了它的帮助，我们现在知道什么lambda collection: max(collection)类型了Callable[[List[int]], int]，但是lambda max_number: -max_number很简单Callable[[int], int]。inflow可以传递任意数量的参数，并且它们可以正常工作。感谢插件。

• RequiresContextFutureResult用于类型功能依赖注入。

FutureResult我们前面讨论 过的over可以用于以函数形式将依赖项显式传递给异步程序。

对未来的计划

在最终发布1.0版之前，我们必须解决一些重要的任务：

• 实现高级类型或它们的仿真（问题）。
• 添加适当的类型类以实现所需的抽象（issue）。
• 也许尝试使用编译器mypyc，这可能会允许将带类型注释的Python程序编译为二进制文件。然后，c代码dry-python/returns将以快几倍的速度运行（issue）。
• 探索用Python编写功能代码的新方法，例如“ do-notation”。

结论

组合和抽象可以解决任何问题。在本文中，我们研究了如何解决功能颜色的问题以及如何编写有效的简单，可读且灵活的代码。并进行类型检查。



尝试使用Dry-Python / Returns并加入“俄罗斯Python周”：在会议上，Dry-Python核心开发人员Pablo Aguilar将举办一个使用Dry-Python编写业务逻辑的研讨会。