深入理解Maybe Monad及其在Python中的实现

from typing import Callable, TypeVar, Generic, Union

# 定义类型变量，用于泛型
T = TypeVar('T')
U = TypeVar('U')

# Just 类：表示一个包含值的Maybe Monad
class Just(Generic[T]):
    def __init__(self, value: T):
        if value is None:
            # 强制Just不包含None，None应该由Nothing处理
            raise ValueError("Just cannot contain None. Use Nothing instead.")
        self.value = value

    def __repr__(self) -> str:
        return f'Just({self.value})'

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Just):
            return NotImplemented
        return self.value == other.value

# Nothing 类：表示一个空值的Maybe Monad
class Nothing:
    # 实现单例模式，因为所有Nothing实例都相同
    def __new__(cls):
        if not hasattr(cls, 'instance'):
            cls.instance = super(Nothing, cls).__new__(cls)
        return cls.instance

    def __repr__(self) -> str:
        return 'Nothing'

    def __eq__(self, other: object) -> bool:
        return isinstance(other, Nothing)

# Maybe 类型别名：表示一个值要么是Just，要么是Nothing
Maybe = Union[Just[T], Nothing]

# bind 函数：Maybe Monad的核心操作
def bind(f: Callable[[U], T], x: Maybe[U]) -> Maybe[T]:
    """
    Maybe Monad的bind操作。
    如果x是Just，则将函数f应用于其内部值，并将结果包装在新的Just中。
    如果x是Nothing，则直接返回Nothing。
    """
    if isinstance(x, Nothing):
        return x
    else: # x 是 Just[U]
        try:
            result = f(x.value)
            # 如果函数f返回None，我们将其视为Nothing
            if result is None:
                return Nothing()
            return Just(result)
        except Exception:
            # 可以在这里添加更复杂的错误处理，例如返回Nothing
            return Nothing()

# unit 函数 (或 return): 将一个普通值提升到Maybe Monad上下文
def unit(value: T) -> Maybe[T]:
    """
    Maybe Monad的unit操作。
    将一个普通值包装成Just Monad。如果值为None，则返回Nothing。
    """
    if value is None:
        return Nothing()
    return Just(value)

# 示例用法
def add_one(n: int) -> int:
    return n + 1

def divide_by_two(n: int) -> Maybe[float]:
    if n == 0:
        return Nothing() # 避免除以零
    return unit(n / 2)

def to_string(x: int) -> str:
    return str(x)

print("--- 单元操作 (unit) ---")
m_one = unit(1)
m_none = unit(None)
print(f"unit(1) -> {m_one}") # Just(1)
print(f"unit(None) -> {m_none}") # Nothing

print("\n--- bind 操作示例 ---")
# 链式操作：Just -> Just -> Just
result1 = bind(add_one, unit(1))
result1 = bind(add_one, result1)
result1 = bind(add_one, result1)
print(f"unit(1) -> add_one -> add_one -> add_one: {result1}") # Just(4)

# 链式操作：Just -> Nothing (通过函数返回Nothing)
result2 = bind(add_one, unit(2))
result2 = bind(divide_by_two, result2) # 2 / 2 = 1.0 -> Just(1.0)
result2 = bind(lambda x: Nothing(), result2) # 强制返回Nothing
result2 = bind(add_one, result2) # Nothing 传播
print(f"unit(2) -> divide_by_two -> Nothing -> add_one: {result2}") # Nothing

# 链式操作：Nothing 传播
result3 = bind(add_one, Nothing())
result3 = bind(add_one, result3)
print(f"Nothing() -> add_one -> add_one: {result3}") # Nothing

# 结合使用 unit 和 bind
print(f"bind(add_one, unit(5)): {bind(add_one, unit(5))}") # Just(6)
print(f"bind(divide_by_two, unit(4)): {bind(divide_by_two, unit(4))}") # Just(2.0)
print(f"bind(divide_by_two, unit(0)): {bind(divide_by_two, unit(0))}") # Nothing

# 链式调用，处理可能为None的中间结果
def get_user_id(user_name: str) -> Maybe[int]:
    if user_name == "Alice":
        return unit(101)
    return Nothing()

def get_user_email(user_id: int) -> Maybe[str]:
    if user_id == 101:
        return unit("alice@example.com")
    return Nothing()

def process_email(email: str) -> Maybe[str]:
    return unit(f"Processed: {email.upper()}")

# 成功获取并处理
user_data_alice = bind(get_user_email, bind(get_user_id, unit("Alice")))
processed_alice_email = bind(process_email, user_data_alice)
print(f"\nAlice's processed email: {processed_alice_email}") # Just('Processed: ALICE@EXAMPLE.COM')

# 失败（用户不存在）
user_data_bob = bind(get_user_email, bind(get_user_id, unit("Bob")))
processed_bob_email = bind(process_email, user_data_bob)
print(f"Bob's processed email: {processed_bob_email}") # Nothing