Zero Boilerplate , Zero Runtime errors : Coding with Monads

Note: this post will be updated with time

API backend programming can be stressful and repetitive. For the past one year, having refactored and rewritten over 10k lines of backend code, I understood the importance of following a proper structure/ discipline while coding backend APIs.
Thankfully, the concept of Monadic programming already existed in languages like Haskell and AngularJS.

This post will show you a template to implement a coding discipline like monads in Python ,which if followed will let you have zero runtime issues and minimal to no boilerplate in your code. For me, this helped me reduce my backend code volume to 1/3rd of its size, enhancing logging and stability, with zero performance latencies added.

Create the monad class
Monads are pure functions. This implicitly means that the type of input it takes in and type of output it produces will be the same. In this case, that state is a dict{}. This also means, that you cannot have exceptions or None type returns in your functions.

Our Monad class is defined to have :
state: This is the current state we have in our monad which represents the data we have currently processed. This is also the argument which we pass to the functions in out monads. We create a mondaic class which takes in a dict as the initial state. For API backend code, this state can be your request parameters. Think of it as if your'e passing the same variables in the function, just that they are wrapped in a dict and the variables names are the keys in the dict.

fns: It has a list of functions, that are function pointers to functions to be executed in order. for example, you can design your sign_up flow as:
[process_request_params, check_username_password, create_token, establish_session]

The .execute will call these functions in a loop till till the final state is reached. The final state should effectively be your response output. Here's what the flow looks like from one state to another state

efn: suppose your code breaks in-between function in the pipeline. The error function efn will process the current state of the monad and output the error as you require in the response. This saves you from getting 500 server errors from backend.

Our monadic code relies on the "status" parameter in status dict, to get "error" reports in-between functions.

class MaybeErrorMonad(object):

    def __init__(self, state, fns, efn):

        self.state = state  # current state (dict) in monadic pipeline
        self.fns = fns      # a list of functions in order of execution
        self.error_fn = efn # error function that reports any error, based on self.state
        self.finish = False # a flag to indicate that the self.state is final state

    def value(self):
        return self.state

    def execute(self):
        # a wrapper class that does execution of functions on current self.state in order
        def _fun():
            if self.finish:
                return self.state
            computation = self.state
            for i in self.fns:
                computation = i(computation)
                if computation.get("status") == "error":
                    return self.error_fn(computation)
            self.state = computation
            self.finish = True
            return self.state

        return _fun()

Next, we define some functions for testing our monad. I have added an error_handler_wrapper wrapper function which will ensure that our functions always return a proper response in a dict , if anything breaks mid-way. You can design this function to include logs with the function name and arguments, with current parameters for better debugging in production code. We also define our custom error function to output our error response , when we get error status midway.

def err_handler_wrapper(fun):
    # a error handler wrapper function to report any execution errors in dict
    def wrapper(*args):
        try:
            state = fun(*args)
        except:
            return {"status": "error", "reason": "Girlfriend not found"}
        return state
    return wrapper

def efn(state):
    return {"status": "Rejected :'(", "reason":state.get('reason') }

@err_handler_wrapper
def get_flowers(state):
    print "got flowers for %s"%state['girlfriend']
    state.update({"flowers": True})
    return state

@err_handler_wrapper
def get_chocolates(state):
    print "got chocolates for %s"%state['girlfriend']
    state.update({"chocolates": True})
    return state

@err_handler_wrapper
def propose(state):
    if state.get('flowers') and state.get('chocolates'):
        return {"status": "Success! she said yes"}
    return {"status": "she said no"}

Lets run our example where a guy tries to propose his crush.
if he buys just flowers, he will get rejected.
if he buys chocolates and flowers, he gets accepted.
if there is no 'girlfriend' in initial state, theres an error.

# CASE 1
to_do = [get_flowers ,get_chocolates, propose]
initial_state = {'girlfriend': 'Alice'}
executor = MaybeErrorMonad(initial_state, to_do, efn)
result = executor.execute()
print "CASE 1: %s \n"%str(result)

 #CASE 2
initial_state = {} # no girlfriend
executor = MaybeErrorMonad(initial_state, to_do, efn)
result = executor.execute()
print "CASE 2: %s \n"%str(result)

# CASE 3
to_do = [get_flowers, propose] # rejection beacuse of not enough efforts :P
initial_state = {'girlfriend': 'Alice'}
executor = MaybeErrorMonad(initial_state, to_do, efn)
result = executor.execute()
print "CASE 3: %s \n"%str(result)

And ...
Here's our output as expected.

This was a fairly simple example. Imagine the power of now being able to execute huge functions in just a simple sequential todo list with minimal branching. TRY IT NOW !!!

In the next update, we will learn how to structure our dict states and how to handle exceptions , and get parameters from proper context of execution.

For more ideas of elimination boilerplate, read this:
Hypercubed!!
Eliminate nested loops

Read more about monads here:-
Monadic Programming
Monads In python: PyMonad