Python 中函数式编程风格的方法链调用

Question

下面是一些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:
    # (get_api_data)(load_db)(notify)(req).else(lambda x -> API_ERROR : print(Error))
    # How to implement this in functional way
    pass


process_request_non_fp(Request(req_id="MY_REQUEST"))

Answer 1

你可以通过使用赋值表达式来避免使用 if-else 结构。

if type(resp = get_api_data(request=req)) is Response and type(api_res := load_db(response=resp)) is ACK and type(ack_resp := notify(api_res)) is bool:
    return ack_resp
return false

另外，在你的代码中，如果已经有了 return，就可以省略多余的 else。

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
     return False

Answer 2

想要实现的效果可以通过使用Either-Monad和柯里化来达到。

import random
from pydantic import BaseModel
from pymonad.tools import curry
from pymonad.either import Either, Left, Right


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


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


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


@curry(1)
def notify(ack: Ack) -> bool | NOTIFICATION_ERROR:
    if random.choice([True, False]):
        print(f" Success !! {ack} Email notification sent")
        return True
    else:
        print("Fail !! Email notification sent")
        return Left(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:
    result = (
        Either.insert(req)
        .then(get_api_data)
        .then(load_db)
        .then(notify)
        .either(
            lambda on_failure: print(f"Error: {on_failure}"),
            lambda on_success: on_success,
        )
    )
    print(result)


process_request_fp(Request(req_id="MY_REQUEST"))