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

Question

下面是简单的Python代码，其中process_request_non_fp方法展示了如何处理IF-ELSE条件的问题（make-api -> load-db -> notification）

我正在尝试以功能方式摆脱 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="[email protected]")
    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

您还可以避免在代码中的

else

之后出现多余的

return

。

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