如何用函数式风格实现Python方法链式调用并处理错误？

以下是一段Python代码，其中process_request_non_fp方法通过嵌套IF-ELSE处理请求流程（调用API→加载数据库→发送通知）。现在需要消除IF-ELSE语句，改用函数式链式调用方式处理，并通过专门方法统一处理错误。可以借助functools、toolz等工具库实现吗？

import random
from pydantic import BaseModel


class APP_ERROR(BaseModel):
    error_message: str


class DB_ERROR(APP_ERROR):
    pass


class API_ERROR(APP_ERROR):
    pass


class NOTIFICATION_ERROR(APP_ERROR):
    pass


class Request(BaseModel):
    req_id: str


class Response(BaseModel):
    response: str


class Ack(BaseModel):
    email: str


def get_api_data(request: Request) -> Response | API_ERROR:
    if random.choice([True, False]):
        print(" Success !! API data successfully retrieved")
        return Response(response="Success")
    else:
        print(" Fail !! API data FAILED")
        return API_ERROR(error_message="API Error")


def load_db(response: Response) -> Ack | DB_ERROR:
    if random.choice([True, False]):
        print(" Success !!  Data loaded to DB")
        return Ack(email="test@gmail.com")
    else:
        print("Fail !! DB Error")
        return DB_ERROR(error_message="DB Error")


def notify(ack: Ack) -> bool | NOTIFICATION_ERROR:
    if random.choice([True, False]):
        print(" Success !! Email notification sent")
        return True
    else:
        print("Fail !! Email notification sent")
        return NOTIFICATION_ERROR(error_message="Notification error")


def process_request_non_fp(req: Request) -> bool:
    resp = get_api_data(request=req)
    if type(resp) is Response:
        api_res = load_db(response=resp)
        if type(api_res) is Ack:
            ack_resp = notify(api_res)
            if type(ack_resp) is bool:
                return ack_resp
            else:
                return False
        else:
            return False
    else:
        return False


def process_request_fp(req: Request) -> bool:
    # 如何用函数式方式实现？
    pass


process_request_non_fp(Request(req_id="MY_REQUEST"))

解决方案：基于Either Monad的函数式链式调用 #

函数式编程中处理这类分支逻辑最常用的是Either Monad模式：用一个容器封装"成功结果"或"错误信息"，通过bind方法实现链式调用——只有当前步骤成功时才执行下一个函数，失败则直接传递错误，全程无需IF-ELSE判断。结合toolz.pipe可以实现简洁的链式流程。

1. 实现轻量Either容器

from typing import Generic, TypeVar, Callable, Optional
from toolz import pipe

# 泛型类型定义
T = TypeVar('T')
E = TypeVar('E')

class Either(Generic[T, E]):
    def __init__(self, is_success: bool, value: Optional[T] = None, error: Optional[E] = None):
        self.is_success = is_success
        self.value = value
        self.error = error

    @classmethod
    def success(cls, value: T) -> 'Either[T, E]':
        return cls(is_success=True, value=value)

    @classmethod
    def failure(cls, error: E) -> 'Either[T, E]':
        return cls(is_success=False, error=error)

    # 链式调用核心方法：成功则执行传入的函数，失败则直接返回自身
    def bind(self, func: Callable[[T], 'Either[any, E]']) -> 'Either[any, E]':
        if self.is_success:
            return func(self.value)
        return self

2. 改造原有函数返回Either类型

把每个步骤的函数修改为返回Either容器，明确区分成功/失败状态：

def get_api_data(request: Request) -> Either[Response, API_ERROR]:
    if random.choice([True, False]):
        print(" Success !! API data successfully retrieved")
        return Either.success(Response(response="Success"))
    else:
        print(" Fail !! API data FAILED")
        return Either.failure(API_ERROR(error_message="API Error"))

def load_db(response: Response) -> Either[Ack, DB_ERROR]:
    if random.choice([True, False]):
        print(" Success !!  Data loaded to DB")
        return Either.success(Ack(email="test@gmail.com"))
    else:
        print("Fail !! DB Error")
        return Either.failure(DB_ERROR(error_message="DB Error"))

def notify(ack: Ack) -> Either[bool, NOTIFICATION_ERROR]:
    if random.choice([True, False]):
        print(" Success !! Email notification sent")
        return Either.success(True)
    else:
        print("Fail !! Email notification failed")
        return Either.failure(NOTIFICATION_ERROR(error_message="Notification error"))

3. 实现函数式链式处理流程

用toolz.pipe按顺序传递请求，通过bind连接各个步骤，自动处理错误分支：

def process_request_fp(req: Request) -> bool:
    # 链式执行：请求→获取API数据→加载数据库→发送通知
    final_result = pipe(
        req,
        get_api_data,
        lambda res: res.bind(load_db),
        lambda res: res.bind(notify)
    )
    # 最终结果：成功则返回True，失败返回False（和原函数行为一致）
    return final_result.is_success and final_result.value is True

4. 测试调用

process_request_fp(Request(req_id="MY_REQUEST"))

方案说明 #

• 全程无IF-ELSE判断：错误分支由Either.bind自动处理，流程逻辑更清晰。
• 扩展性强：新增步骤只需添加对应函数并在链式调用中增加bind即可。
• 错误可追溯：每个Either容器保存了具体错误信息，可根据需求添加统一错误日志处理。

内容的提问来源于stack exchange，提问作者M80