(a -> b) -> парсер a -> парсер b 

Ага! очень похоже на карту . Действительно, существует обобщенная версия map , которую мы видели ранее ( lift1 ), и мы ограничиваем шаблон, объявляя Parser экземпляром Functor класса типов 9.0003

 > экземпляр Functor Parser, где
> fmap f p = do x <- p
> возврат (ф х) 

после чего мы можем переписать

 > oneInt :: Parser Int
> oneInt = прочитать `fmap` mmanyP digitChar 

Давай попробуем

 ghci> doParse oneInt "123a"
[(123, "а")] 

Разбор арифметических выражений

Давайте воспользуемся вышеизложенным для создания небольшого калькулятора, который анализирует и вычисляет арифметические выражения. По сути, выражение представляет собой либо двоичный операнд, применяемый к двум подвыражениям, либо целое число. Мы можем сформулировать это как

 > calc0 :: Parser Int
> calc0 = binExp <|> oneInt
> где
> binExp = do x <- oneInt
> о <- intOp
> у <- расч0
> вернуть $ x `o` y 

Это работает очень хорошо!

 ghci> doParse calc0 "1+2+33"
[(36,"")]
ghci> doParse calc0 "11+22-33"
[(0,"")] 

, но с минусом

 ghci> doParse calc0 "11+22-33+45"
[(-45,"")] 

А? Что ж, если вы снова посмотрите на код, вы поймете, что приведенное выше было проанализировано как

 11 + ( 22 - (33 + 45)) 

, потому что в каждом binExp мы требуем, чтобы левый операнд был целым числом. Другими словами, мы предполагаем, что каждый оператор является правой ассоциативностью , отсюда и полученный выше результат.

Интересно, можем ли мы попытаться это исправить, просто перевернув заказ

 > calc1 :: Parser Int
> calc1 = binExp <|> oneInt
> где
> binExp = do x <- calc1
> о <- intOp
> y <- oneInt
> вернуть $ x `o` y 

ВИКТОРИНА

Что оценивает следующее?

 ghci> doParse calc1 "11+22-33+45" 

1. [( 11 , "+22-33+45")]
2. [( 33 , "-33+45")]
3. [( 0, "+45")]
4. [( 45 , "")]
5. Ничего из вышеперечисленного

Действительно, здесь есть ошибка… Вы можете это понять?

Подсказка: что вернет следующее?

 ghci> doParse calc1 "2+2" 

Хуже того, у нас нет приоритета, и поэтому

 ghci> doParse calc0 "10*2+100"
[(1020,"")] 

, так как строка анализируется как

 10*(2+100) 

Приоритет

Мы можем добавить как ассоциативность, так и приоритет, разделив синтаксический анализатор на разные уровни. Здесь давайте разделим наши операции на сложение-

 > addOp = плюс `выберите P` минус
> где
> плюс = символ '+' >> возврат (+)
> минус = символ '-' >> возврат (-) 

и приоритет умножения.

 > mulOp = деление на `chooseP`
> где
> раз = символ '*' >> возврат (*)
> разделить = символ '/' >> вернуть div 

Теперь мы можем разделить наш язык на (взаимно рекурсивные) подъязыки, где каждое выражение верхнего уровня анализируется как сумма произведений

 > sumE = addE <|> prodE
> где
> addE = сделать x <- prodE
> о <-допОп
> у <- суммаE
> вернуть $ x `o` y
>
> prodE = mulE <|> factorE
> где
> mulE = do x <- factorE
> о <- mulOp
> у <- prodE
> вернуть $ x `o` y
>
> factorE = parenP sumE <|> oneInt 

Мы можем запустить это

 ghci> doParse sumE "10*2+100"
[(120,"")]
ghci> doParse sumE "10*(2+100)"
[(1020,"")] 

Вы понимаете, почему первый синтаксический анализ вернул 120 ? Что произойдет, если мы поменяем местами порядок prodE и sumE в теле addE (или factorE и prodE в теле prodE )? Почему?

ВИКТОРИНА

Напомним, что в приведенном выше

 factorE :: Parser Int
factorE = parenP sumE <|> oneInt 

Какой тип parenP ?

1. Parser Int
2. Парсер а -> Парсер а
3. а -> парсер
4. Парсер а ->
5. Parser Int -> Parser a

Давайте напишем parenP

 > parenP p = do char '('
> х <- р
> символ ')'
> возврат х 

Шаблон синтаксического анализа: Цепочка

Нет особого смысла злорадствовать по поводу комбинаторов, если мы собираемся писать код, подобный приведенному выше — корпуса sumE и prodE практически идентичны!

Давайте посмотрим на них поближе. В сущности, sumE имеет вид

 prodE + < prodE + < prodE + ... < prodE >>> 

, то есть мы продолжаем связывать вместе значения prodE и добавлять их столько, сколько сможем. Аналогично prodE имеет вид

 фактор E * < фактор E * < фактор E * ... < фактор E >>> 

, где мы продолжаем связывать значения factorE и умножать их столько, сколько сможем. В вышесказанном есть что-то неприятное: операторы сложения правоассоциативны

 ghci> doParse sumE "10-1-1"
[(10,"")] 

Ух! Надеюсь, вы понимаете, почему: это потому, что приведенное выше было проанализировано как 10 - (1 - 1) (правоассоциативный), а не (10 - 1) - 1 (левоассоциативный). У вас может возникнуть соблазн исправить это, просто поменяв местами prodE и sumE

 sumE = addE <|> prodE
  куда
    addE = сделать x <- sumE
              о <- addOp
              у <- prodE
              вернуть $ x `o` y 

, но это было бы катастрофой. Вы видите, почему?

Парсер для sumE напрямую (рекурсивно) вызывает сам себя без каких-либо входных данных! Таким образом, он уходит в глубокий конец и никогда не возвращается. Вместо этого мы хотим убедиться, что мы продолжаем потреблять значения prodE и суммировать их (скорее, как складывать), и поэтому мы можем сделать

 > sumE1 = prodE1 >>= addE1
> где
> addE1 x = захватить x <|> вернуть x
> захватить x = сделать о <- addOp
> у <- prodE1
> addE1 $ x `o` y
>
> prodE1 = factorE1 >>= mulE1
> где
> mulE1 x = захватить x <|> вернуть x
> схватить x = сделать о <- mulOp
> у <- факторE1
> mulE1 $ x `o` y
>
> factorE1 = parenP sumE1 <|> oneInt 

Легко проверить, что приведенное выше действительно ассоциативно слева.

 ghci> doParse sumE1 "10-1-1"
[(8,"")] 

, а также очень легко определить и заключить шаблон вычисления цепочки: единственными отличиями являются парсер base ( prodE1 vs factorE1 ) и бинарная операция ( addOp vs mulOp ). Мы просто вносим эти параметры в наш комбинатор

 > p `chainl` pop = p >>= rest
> где
> отдых x = захватить x <|> вернуть x
> схватить x = сделать о <- поп
> у <- р
> отдых $ x `o` y 

после чего мы можем переписать грамматику в три строки

 > sumE2 = prodE2 `chainl` addOp
> prodE2 = factorE2 `chainl` mulOp
> factorE2 = parenP sumE2 <|> oneInt 

 ghci> doParse sumE2 "10-1-1"
[(8,"")]
ghci> doParse sumE2 "10*2+1"
[(21,"")]
ghci> doParse sumE2 "10+2*1"
[(12,"")] 

На этом мы завершаем наше изучение монадического разбора в классе. Это всего лишь верхушка айсберга. Хотя синтаксический анализ — очень старая проблема, и ее изучали на заре вычислительной техники, мы увидели, как монады привносят свежий взгляд, который недавно был перенесен из Haskell во многие другие языки. Недавно было опубликовано несколько захватывающих статей на эту тему, которые вы можете изучить самостоятельно. Наконец, Haskell поставляется с несколькими библиотеками комбинаторов синтаксических анализаторов, включая Parsec, с которым вы поэкспериментируете в HW2.

λ >

возвратов · PyPI

Сделайте так, чтобы ваши функции возвращали что-то осмысленное, типизированное и безопасное!

Возможности

• Привносит функциональное программирование в мир Python
• Предоставляет набор примитивов для написания декларативной бизнес-логики
• Обеспечивает лучшую архитектуру
• Полностью напечатано с аннотациями и проверено с помощью mypy , совместимо с PEP561
• Добавляет поддержку эмулируемых типов высшего порядка
• Обеспечивает типобезопасные интерфейсы для создания собственных типов данных с применением законов
• Имеет кучу помощников для лучшей композиции
• Pythonic и приятно писать и читать 🐍
• Функции поддержки и сопрограммы, не зависящие от фреймворка
• Легко начать: содержит множество документов, тестов и учебных пособий

Быстрый старт прямо сейчас!

Установка

 pip install возвращает
 

Также необходимо настроить mypy правильно и устанавливаем наш плагин чтобы исправить эту существующую проблему:

 # В setup. cfg или mypy.ini:
[mypy]
плагины =
  возвращает.contrib.mypy.returns_plugin
 

Мы также рекомендуем использовать те же настройки mypy , которые мы используем.

Убедитесь, что вы знаете, как начать работу, ознакомьтесь с нашей документацией! Попробуйте нашу демоверсию.

Содержимое

• Возможно, контейнер, позволяющий писать Нет -бесплатный код
• Контейнер RequiresContext, который позволяет использовать внедрение типизированной функциональной зависимости
• Контейнер результатов, позволяющий избавиться от исключений
• Контейнер ввода-вывода и IOResult, который отмечает все нечистые операции и структурирует их
• Контейнер Future и FutureResult для работы с async code
• Напиши свой собственный контейнер! У вас по-прежнему будут все функции для ваших собственных типов (включая полное повторное использование существующего кода и безопасность типов)

Возможно контейнер

Нет называют самой большой ошибкой в ​​истории информатики.

Итак, что мы можем сделать, чтобы проверить наличие None в наших программах? Вы можете использовать встроенный необязательный тип и напишите много если какое-то нет None: условий. Но с нулевыми проверками здесь и там делает ваш код нечитаемым .

 пользователь: Необязательно[Пользователь]
Discount_program: Необязательно['DiscountProgram'] = Нет
если пользователь не None:
     баланс = user.get_balance()
     если баланс не None:
         кредит = баланс.credit_amount()
         если кредит не None и кредит > 0:
             Discount_program = выберите_скидку (кредит)
 

Или вы можете использовать Может контейнер! Он состоит из типов Some и Nothing , представляющие существующее состояние и пустое (вместо None ) состояние соответственно.

 от ввода импорта Необязательно
из return.maybe импорт
@maybe # декоратор для преобразования существующего Optional[int] в Maybe[int]
def bad_function() -> Дополнительно[int]:
    . ..
возможно_число: Возможно[плавающая] = плохая_функция().bind_Optional(
    лямбда число: число / 2,
)
# => Maybe вернет Some[float], только если есть значение, отличное от None
# В противном случае ничего не вернется
 

Вы можете быть уверены, что метод .bind_Optional() не будет вызываться для Ничего . Забудьте об ошибках, связанных с None , навсегда!

Мы также можем привязать Необязательную функцию возврата к контейнеру. Для этого мы будем использовать метод .bind_Optional .

И вот как будет выглядеть ваш первоначальный рефакторинговый код:

 пользователь: Необязательный[Пользователь]
# Подсказка здесь необязательна, она только помогает читателю:
Discount_program: Возможно['DiscountProgram'] = Возможно.from_необязательно(
    пользователь,
).bind_Optional( # Это не будет вызываться, если `user is None`
    лямбда реальный_пользователь: real_user.get_balance(),
).bind_Optional( # Это не будет вызываться, если `real_user. get_balance()` равно None
    лямбда-баланс: balance.credit_amount(),
).bind_Optional( # И так далее!
    лямбда-кредит: Choose_discount(кредит), если кредит> 0 иначе нет,
)
 

Гораздо лучше, не правда ли?

Контейнер RequiresContext

Многие разработчики используют некоторые виды внедрения зависимостей в Python. И обычно это основано на идее что есть какой-то контейнер и процесс сборки.

Функциональный подход намного проще!

Представьте, что у вас есть игра на основе django , где вы начисляете пользователям очки за каждую угаданную букву в слове (неугаданные буквы помечаются как '.' ):

 из django.http импорт HttpRequest, HttpResponse
из words_app.logic импорта calculate_points
представление защиты (запрос: HttpRequest) -> HttpResponse:
    user_word: str = request.POST['word'] # просто пример
    баллы = calculate_points (пользовательское_слово)
    ... # позже вы каким-то образом показываете результат пользователю
# Где-то в вашем `words_app/logic. py`:
def calculate_points (word: str) -> int:
    угаданный_letters_count = len([буква вместо буквы в слове, если буква != '.'])
    вернуть _award_points_for_letters (угаданное_число_букв)
def _award_points_for_letters (предположительно: int) -> int:
    вернуть 0, если угадано < 5, иначе угадано # возможно минимум 6 баллов!
 

Отлично! Это работает, пользователи довольны, ваша логика чиста и удивительна. Но позже вы решаете сделать игру более увлекательной: давайте сделаем минимальный порог ответственных писем настраивается для дополнительной задачи.

Вы можете просто сделать это напрямую:

 def _award_points_for_letters(угадано: int, threshold: int) -> int:
    вернуть 0, если угадано < порога, иначе угадано
 

Проблема в том, что _award_points_for_letters глубоко вложены. И тогда вы должны пройти порог через весь стек вызовов, включая calculate_points и все другие функции, которые могут быть в пути. Все они должны будут принимать в качестве параметра порог ! Это совсем не полезно! Большие кодовые базы будут сильно страдать от этого изменения.

Хорошо, вы можете напрямую использовать django.settings (или аналогичный) в вашей функции _award_points_for_letters . И разрушает вашу чистую логику специфическими для фреймворка деталями . Это некрасиво!

Или вы можете использовать контейнер RequiresContext . Посмотрим, как изменится наш код:

 из настроек импорта django.conf
из django.http импортировать HttpRequest, HttpResponse
из words_app.logic импорта calculate_points
представление защиты (запрос: HttpRequest) -> HttpResponse:
    user_word: str = request.POST['word'] # просто пример
    points = calculate_points(user_words)(settings) # передача зависимостей
    ... # позже вы каким-то образом показываете результат пользователю
# Где-то в вашем `words_app/logic.py`:
из протокола импорта typing_extensions
из импорта return. context RequiresContext
class _Deps(Protocol): # мы полагаемся на абстракции, а не на прямые значения или типы
    WORD_THRESHOLD: целое
def calculate_points(word: str) -> RequiresContext[int, _Deps]:
    угаданный_letters_count = len([буква вместо буквы в слове, если буква != '.'])
    вернуть _award_points_for_letters (угаданное_число_букв)
def _award_points_for_letters (предположительно: int) -> RequiresContext[int, _Deps]:
    вернуть RequiresContext(
        lambda deps: 0, если угадано, < deps.WORD_THRESHOLD, иначе угадано,
    )
 

И теперь вы можете передавать свои зависимости действительно прямым и явным образом. И имейте тип-безопасность, чтобы проверить, что вы передаете, чтобы прикрыть спину. Ознакомьтесь с документацией RequiresContext для получения дополнительной информации. Там вы узнаете, как сделать '.' также настраивается.

У нас также есть RequiresContextResult для операций, связанных с контекстом, которые могут завершиться ошибкой. А также RequiresContextIOResult и RequiresContextFutureResult.

Контейнер результатов

Убедитесь, что вы также знаете Железнодорожное программирование.

Простой подход

Рассмотрим этот код, который вы можете найти в любом проекте Python .

 запросов на импорт
def fetch_user_profile(user_id: int) -> 'UserProfile':
    """Выбирает профиль пользователя из внешнего API."""
    ответ = запросы.get('/api/users/{0}'.format(user_id))
    ответ.raise_for_status()
    вернуть ответ.json()
 

Кажется законным, не так ли? Это также кажется довольно простым кодом для тестирования. Все, что вам нужно, это издеваться над request.get , чтобы вернуть нужную структуру.

Но в этом крошечном примере кода есть скрытые проблемы которые почти невозможно заметить с первого взгляда.

Скрытые проблемы

Взглянем на точно такой же код, но со всеми скрытыми проблемами объяснил.

 запросов на импорт
def fetch_user_profile(user_id: int) -> 'UserProfile':
    """Выбирает профиль пользователя из внешнего API. """
    ответ = запросы.get('/api/users/{0}'.format(user_id))
    # Что если мы попытаемся найти несуществующего пользователя?
    # Или сеть выйдет из строя? Или сервер вернет 500?
    # В этом случае следующая строка завершится ошибкой с исключением.
    # Нам нужно обработать все возможные ошибки в этой функции
    # и не возвращать поврежденные данные потребителям.
    ответ.raise_for_status()
    # Что делать, если мы получили неверный JSON?
    # Следующая строка вызовет исключение!
    вернуть ответ.json()
 

Теперь все (возможно, все?) проблемы решены. Как мы можем быть уверены, что эта функция будет безопасной использовать внутри нашей сложной бизнес-логики?

Мы действительно не можем быть уверены! Нам нужно будет создать лот из попробовать и кроме случаев просто чтобы поймать ожидаемые исключения. Со всем этим бардаком наш код станет сложным и нечитаемым!

Или мы можем использовать верхний уровень , кроме Исключения: случай поймать буквально все. И таким образом мы закончим тем, что поймаем нежелательных. Такой подход может надолго скрыть от нас серьезные проблемы.

Пример трубы

 запросы на импорт
from return.result import Результат, безопасный
из потока импорта return.pipeline
привязка импорта из return.pointfree
def fetch_user_profile(user_id: int) -> Результат['UserProfile', Exception]:
    """Выбирает `UserProfile` TypedDict из стороннего API."""
    обратный поток(
        Идентификатор пользователя,
        _Сделать запрос,
        привязать (_parse_json),
    )
@Безопасно
def _make_request (user_id: int) -> запросы. Ответ:
    # TODO: мы еще не закончили с этим примером, читайте больше о `IO`:
    ответ = запросы.get('/api/users/{0}'.format(user_id))
    ответ.raise_for_status()
    вернуть ответ
@Безопасно
def _parse_json (ответ: запросы. Ответ) -> «Профиль пользователя»:
    вернуть ответ.json()
 

Теперь у нас есть чистый, безопасный и декларативный способ чтобы выразить наши потребности бизнеса:

• Мы начинаем с запроса, который может не получиться в любой момент,
• Затем анализ ответа, если запрос был успешным,
• А затем вернуть результат.

Теперь вместо возврата обычных значений мы возвращаем значения, завернутые в специальный контейнер благодаря @Безопасно декоратор. Он вернет Success[YourType] или Failure[Exception]. И никогда не будет бросать нам исключение!

Мы также используем поток и связать функции для удобной и декларативной композиции.

Таким образом, мы можем быть уверены, что наш код не взломает случайные места из-за некоторого неявного исключения. Теперь мы контролируем все детали и готовы к явным ошибкам.

Мы еще не закончили с этим примером, давайте продолжим улучшать его в следующей главе.

IO-контейнер

Давайте посмотрим на наш пример под другим углом. Все его функции выглядят как обычные: невозможно сказать, чистые ли они или нечистым с первого взгляда.

Это приводит к очень важному следствию: мы начинаем смешивать чистый и нечистый код вместе . Мы не должны этого делать!

Когда эти два понятия смешаны мы очень сильно страдаем при тестировании или повторном использовании. Почти все должно быть чистым по умолчанию. И мы должны явно помечать нечистые части программы.

Вот почему мы создали контейнер IO чтобы пометить нечистые функции, которые никогда не выходят из строя.

Эти нечистые функции используют random , текущая дата и время, среда или консоль:

 случайный импорт
импортировать дату и время как dt
из return.io импортировать IO
def get_random_number() -> IO[int]: # или используйте декоратор `@impure`
    return IO(random.randint(1, 10)) # не является чистым, потому что случайный
сейчас: Callable[[], IO[dt.datetime]] = impure(dt.datetime.now)
@нечистый
def return_and_show_next_number (предыдущий: int) -> int:
    следующий_номер = предыдущий + 1
    print(next_number) # не является чистым, потому что IO
    вернуть следующий_номер
 

Теперь мы ясно видим, какие функции чистые, а какие нечистые. Это очень помогает нам в создании больших приложений, модульном тестировании вашего кода, и совместное составление бизнес-логики.

Troublesome IO

Как уже было сказано, мы используем IO , когда обрабатываем функции, которые не дают сбоев.

Что, если наша функция может дать сбой и окажется нечистой? Подобно request.get() у нас было ранее в нашем примере.

Тогда мы должны использовать специальный IOResult 9Тип 0012 вместо обычного Результат . Давайте найдем разницу:

• Наша функция _parse_json всегда возвращает тот же результат (надеюсь) для того же ввода: вы можете либо проанализировать действительный json , либо потерпеть неудачу на недопустимом. Вот почему мы возвращаем чистый Result , внутри
нет IO
• Наша функция _make_request нечиста и может дать сбой. Попробуйте отправить два одинаковых запроса с подключением к Интернету и без него. Результат будет другим для одного и того же ввода. Вот почему мы должны использовать IOResult здесь: он может выйти из строя и имеет IO

Итак, чтобы выполнить наше требование и отделить чистый код от нечистого, мы должны реорганизовать наш пример.

Явный ввод-вывод

Давайте сделаем наш ввод-вывод явный!

 запросов на импорт
из return.io импортировать IOResult, impure_safe
из return.result безопасно импортировать
из потока импорта return.pipeline
из return.pointfree импортировать bind_result
def fetch_user_profile(user_id: int) -> IOResult['UserProfile', Exception]:
    """Выбирает `UserProfile` TypedDict из стороннего API."""
    обратный поток(
        Идентификатор пользователя,
        _Сделать запрос,
        # до: def (Ответ) -> UserProfile
        # после сейфа: def (Response) -> ResultE[UserProfile]
        # после bind_result: def (IOResultE[Response]) -> IOResultE[UserProfile]
        результат связывания (_parse_json),
    )
@impure_safe
def _make_request (user_id: int) -> запросы. Ответ:
    ответ = запросы.get('/api/users/{0}'.format(user_id))
    ответ.raise_for_status()
    вернуть ответ
@Безопасно
def _parse_json (ответ: запросы. Ответ) -> «Профиль пользователя»:
    вернуть ответ. json()
 

И позже мы можем использовать unsafe_perform_io где-то на верхнем уровне нашей программы, чтобы получить чистое (или «настоящее») значение.

В результате этого сеанса рефакторинга мы знаем все о нашем коде:

• Какие части могут выйти из строя,
• Какие части нечисты,
• Как составить их умным, удобочитаемым и типобезопасным способом.

Будущий контейнер

Есть несколько проблем с асинхронным кодом в Python:

1. Вы не можете вызвать асинхронную функцию из синхронной
2. Любое неожиданно созданное исключение может разрушить весь цикл обработки событий
3. Уродливая композиция с лотами await утверждений
Контейнеры

Future и FutureResult решают эти проблемы!

Смешивание синхронного и асинхронного кода

Основная особенность Future заключается в том, что он позволяет запускать асинхронный код при сохранении контекста синхронизации. Давайте посмотрим пример.

Допустим, у нас есть две функции, первый возвращает число, а второй увеличивает его:

 async def first() -> int:
    вернуть 1
def second(): # Как мы можем вызвать `first()` отсюда?
    return first() + 1 # Бум! Не делай этого. Мы иллюстрируем проблему здесь.
 

Если мы попытаемся просто запустить first() , мы просто создадим нежданную сопрограмму. Он не вернет нужное нам значение.

Но если мы попытаемся запустить await first() , тогда нам нужно будет изменить second на async . А иногда это невозможно по разным причинам.

Однако с Future мы можем «притвориться», что вызываем асинхронный код из кода синхронизации:

 из return.future import Future
def second() -> Future[int]:
    вернуть будущее (первый ()). карта (лямбда число: число + 1)
 

Не касаясь нашей первой асинхронной функции или сделав секунд асинхронными, мы достигли своей цели. Теперь наше асинхронное значение увеличивается внутри функции синхронизации.

Однако Future по-прежнему требует выполнения внутри надлежащего цикла обработки событий:

 import anyio # или asyncio, или любая другая библиотека
# Затем мы можем передать наше `Future` любой библиотеке: asyncio, trio, curio.
# И использовать любой цикл событий: обычный, uvloop, даже собственный и т.д.
утверждать anyio.run(second().awaitable) == 2
 

Как видите, Future позволяет вам для работы с асинхронными функциями из контекста синхронизации. И смешать эти две сферы вместе. Используйте необработанный Будущее для операций, которые не могут завершиться ошибкой или вызвать исключения. Практически та же логика была у нас с контейнером IO .

Асинхронный код без исключений

Мы уже рассмотрели, как работает Результат как для чистого, так и для нечистого кода. Основная идея такова: мы не вызываем исключения, мы их возвращаем. Это , особенно критично в асинхронном коде, потому что одно исключение может разрушить все наши сопрограммы работают в одном цикле событий.

У нас есть удобная комбинация из Future и Result контейнеры: FutureResult . Опять же, это точно так же, как IOResult , но для нечистого асинхронного кода. Используйте его, когда у вашего Future могут возникнуть проблемы: например HTTP-запросы или операции с файловой системой.

Вы можете легко превратить любую дикую сопрограмму в спокойную FutureResult :

 import anyio
из return.future импортировать future_safe
из return.io импортировать IOFailure
@future_safe
асинхронное повышение защиты():
    поднять ValueError('Не так быстро!')
ioresult = anyio.run(raising.awaitable) # все контейнеры `Future` return IO
assert ioresult == IOFailure(ValueError('Не так быстро!')) # True
 

Использование FutureResult защитит ваш код от исключений. Вы всегда можете await или выполнить внутри цикла событий любой FutureResult чтобы получить экземпляр IOResult синхронизации для работы с ним в режиме синхронизации.

Лучшая асинхронная композиция

Раньше вам приходилось делать довольно много await ing при написании async code:

 async def fetch_user(user_id: int) -> 'User':
    ...
async def get_user_permissions (пользователь: «Пользователь») -> «Разрешения»:
    ...
async def sure_allowed (разрешения: «Разрешения») -> логическое значение:
    ...
async def main(user_id: int) -> bool:
    # Кроме того, не забудьте обработать все возможные ошибки с помощью `try/except`!
    user = await fetch_user(user_id) # Мы будем ждать каждый раз, когда будем использовать coro!
    разрешения = ожидание get_user_permissions (пользователь)
    возврат ожидания гарантировать_разрешение (разрешения)
 

Некоторым это нравится, но некоторым этот императивный стиль не нравится. Проблема в том, что выбора не было.

Но теперь вы можете сделать то же самое в функциональном стиле! С помощью контейнеров Future и FutureResult :

 import anyio
из return.future импортировать FutureResultE, future_safe
из return.io импортировать IOSuccess, IOFailure
@future_safe
async def fetch_user(user_id: int) -> 'Пользователь':
    ...
@future_safe
async def get_user_permissions (пользователь: «Пользователь») -> «Разрешения»:
    ...
@future_safe
async def sure_allowed (разрешения: «Разрешения») -> логическое значение:
    ...
def main(user_id: int) -> FutureResultE[bool]:
    # Теперь мы можем превратить main в функцию синхронизации, она вообще не ожидает.
    # Мы также больше не заботимся об исключениях, они уже обработаны.
    вернуть fetch_user(user_id).bind(get_user_permissions).bind(ensure_allowed)
correct_user_id: int # имеет необходимые разрешения
Banned_user_id: int # не имеет необходимых разрешений
неправильно_user_id: int # не существует
# Мы можем получить правильные бизнес-результаты:
утверждать anyio. run(main(correct_user_id).awaitable) == IOSuccess(True)
утверждать anyio.run(main(banned_user_id).awaitable) == IOSuccess(False)
# Или у нас могут быть ошибки по пути:
утверждать anyio.run(main(wrong_user_id).awaitable) == IOFailure(
    UserDoesNotExistError(...),
)
 

Или даже что-то действительно причудливое:

 из return.pointfree import bind
из потока импорта return.pipeline
def main(user_id: int) -> FutureResultE[bool]:
    обратный поток(
        fetch_user(user_id),
        привязать (get_user_permissions),
        привязать (обеспечить_разрешение),
    )
 

Позже мы также можем реорганизовать наши логические функции для синхронизации и вернуть FutureResult .

Красиво, не так ли?

Еще!

Хотите больше? Перейти к документам! Или прочитайте эти статьи:

• Исключения Python считаются анти-шаблоном
• Применение принципа единой ответственности в Python
• Внедрение типизированной функциональной зависимости в Python
• Какой должна быть асинхронность
• Высшие родственные типы в Python
• Сделайте тесты частью вашего приложения

У вас есть статья для отправки? Не стесняйтесь открывать запрос на вытягивание!

— ⭐️ —

Drylabs поддерживает dry-python и помогает тем, кто хочет использовать его в своих организациях.