В последние несколько лет ключевое слово async
и семантика асинхронного программирования проникла во многие популярные языки программирования: JavaScript, Rust, C#, и многие другие. Конечно, в Python тоже есть async/await
, они появились в Python 3.5.
В этой статье хочу обсудить проблемы асинхронного кода, порассуждать об альтернативах и предложить новый подход поддерживать и синхронные, и асинхронные приложения одновременно.
async
все еще поддерживается далеко не так хорошо, как синхронные возможности Python.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
— тот же самый код автоматически начинает работать асинхронно.dry-python/returns
уже есть правильные абстракции.pip install returns httpx anyio
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()
. Успешные выходы при этом оборачиваются в тип 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
.@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
и выполняет его. В то время как в асинхронном варианте требуется event-loop, как для обычной корутины. 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")
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>>
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
from returns.pipeline import flow
assert flow(
[1, 2, 3],
lambda collection: max(collection),
lambda max_number: -max_number,
) == -3
Any
. Вывод mypy
решает эту проблему.lambda collection: max(collection)
типа Callable[[List[int]], int]
, а lambda max_number: -max_number
просто Callable[[int], int]
. Во flow
можно передать любое количество аргументов, и все они будут отлично работать. Всё благодаря плагину.FutureResult
, о которой мы говорили ранее, можно использовать для того, чтобы явно передать зависимости в асинхронные программы в функциональном стиле.mypyc
, что потенциально позволит компилировать типизированные аннотированные Python-программы в двоичный файл. Тогда код с dry-python/returns
будет работать в несколько раз быстрее (issue).К сожалению, не доступен сервер mySQL