Python method chaining in functional programming style

Question

Below is Python code, where the process_request_non_fp method shows how to handle the problem with IF-ELSE condition (make-api -> load-db -> notify).

I'm trying to get rid of IF-ELSE and chain in functional way with ERROR being handled in special method.

Is there any helper from functools, toolz etc to achieve this in simple functional way?

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

The desired behavior can be obtained using Either-Monad and currying

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="[email protected]")
    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"))

Answer 2

You can avoid if-else using assignment expression.

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

Also you can just avoid redundant else after return in your code.

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