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="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"))
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="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"))