首页 > 解决方案 > 在 Python 中,为什么不使用不同的 `bind` 方法而不是用于“无短路”输入验证的应用函子呢?

问题描述

我一直在探索如何使用 monad/applicative functor 从头开始​​在 Python 中进行数据验证。

特别是,我创建了一个Validation允许表示验证结果的新类:SuccessError. 它基本上是一个标称Result类型,但不完全是:

from __future__ import annotations

from typing import List, TypeVar, Generic, Callable, Optional

A = TypeVar('A')
B = TypeVar('B')


class Validation(Generic[A]):
    def __init__(self, value: Optional[A], errors: List[str]) -> None:
        self.value = value
        self.errors = errors

    def is_success(self):
        raise NotImplementedError('')

    def is_error(self):
        return not self.is_success()

    def map(self, function: Callable[[A], B]) -> Validation[B]:
        raise NotImplementedError('')

    def bind(self, function: Callable[[A], Validation[B]]) -> Validation[B]:
        raise NotImplementedError('')


class Error(Validation[A]):
    def __init__(self, errors: List[str]) -> None:
        super().__init__(None, errors)

    def is_success(self):
        return False

    def map(self, function: Callable[[A], B]) -> Validation[B]:
        return self

    def bind(self, function: Callable[[A], Validation[B]]) -> Validation[B]:
        return self


class Success(Validation[A]):
    def __init__(self, value: A) -> None:
        super().__init__(value, [])

    def is_success(self):
        return True

    def map(self, function: Callable[[A], B]) -> Validation[B]:
        return function(self.value)

    def bind(self, function: Callable[[A], Validation[B]]) -> Validation[B]:
        return function(self.value)

这工作得很好,它会在遇到第一个错误时停止:

def check_1(data: str) -> Validation[str]:
    print(f"data 1: {data}")
    return Success(data)


def check_2(data: str) -> Validation[str]:
    print(f"data 2: {data}")
    return Error(['unable to validate with check 2'])


def check_3(data: str) -> Validation[str]:
    print(f"data 3: {data}")
    return Error(['unable to validate with check 3'])

if __name__ == "__main__":
    print("Stopping at the first error (aka. monad's bind):")
    checks = Success('input to check') \
        .bind(check_1) \
        .bind(check_2) \
        .bind(check_3)
    assert checks.value is None
    assert checks.errors == ['unable to validate with check 2']

为什么要bind这样定义?

没有什么能阻止我定义一个 new bind2,它不会在遇到第一个错误时停止,而是尝试所有错误直到结束:

class Validation(Generic[A]):
    # (...)

    def bind2(self, function: Callable[[A], Validation[B]]) -> Validation[B]:
        raise NotImplementedError('')

class Error(Validation[A]):
    # (...)
    def bind2(self, function: Callable[[A], Validation[B]]) -> Validation[B]:
        return self


class Success(Validation[A]):
    # (...)
    def bind2(self, function: Callable[[A], Validation[B]]) -> Validation[B]:
        application = function(self.value)
        if application.is_success():
            return application
        else:
            self.errors += application.errors
            return self

这正如预期的那样工作:

if __name__ == "__main__":
    print("Showing all errors without stopping at the first one:")
    checks = Success('input to check') \
        .bind2(check_1) \
        .bind2(check_2) \
        .bind2(check_3)
    assert checks.value == 'input to check'
    assert checks.errors == ['unable to validate with check 2', 'unable to validate with check 3']

鉴于此,为什么不进行bind2不短路呢?

我读过apply应该在这里使用应用函子的方法,但我真的不明白这有什么帮助,因为它的签名是def apply(self, function: Validation[Callable[[A], B]]) -> Validation[B]:(如果我没记错的话)不接受我的任何check_x函数签名。

标签: pythonfunctional-programmingapplicative

解决方案


推荐阅读