Monad Example: Sequencing API Calls

In the previous two articles we saw what monads are and an example of using them in the robot kata. If you haven't read about monads, take a look at the monad introduction as we will be building on that.

Introducing Monads in Functional Programming

Monads have a reputation as being a very complicated aspect of functional programming. In this article we demystify them and learn how to apply monads in our code

Playful PythonSiddharta Govindaraj

In this article, we will look at a fairly common scenario that all of us have experienced – implementing logic that makes multiple API calls.

Problem Statement

In this problem, we want to find out the country that the user is from. This is a two step process:

• Make an API call to https://api.ipify.org/?format=json to find out the IP address
• Use that IP address to make a call to http://ipwho.is/{ip} which will return the country that the IP belongs to

In case the second call to get the country fails for some reason, then we have a backup flow:

• Make a call to https://ipinfo.io/{ip}/geo  which will return an ISO 2-letter country code
• Take that 2-letter code and call https://countrycode.dev/api/countries/iso2/{country_code} to get the country name

Here we have a sequence of API calls that we need to make. The challenge is that any of these calls could fail and we need to handle that.

When we run the script, the code will run this sequence and print out the country. If the call to ipwho.is fails, we can still try again through the backup flow. If the backup flow also fails then we just print out an error saying which call failed.

The Result Monad

This time, apart from success / failure, we also need to store the error message in case of failure. This kind of monad which stores additional information about the failure is called the Result monad (Result is the terminology from F#, in Haskell it is called Either)

Just like the Maybe monad, the Result monad has two states

The implementation is similar to the Maybe monad from the previous article, except the Error state also takes an input.

class Ok:
    def __init__(self, value):
        self.value = value

    def map(self, fn):
        return Ok(fn(self.value))

    def flatmap(self, fn):
        return fn(self.value)

class Error:
    def __init__(self, error):
        self.error = error

    def map(self, fn):
        return self

    def flatmap(self, fn):
        return self

Implementing the API calls

We are now ready to implement the API calls. We will use the requests module to make the calls.

We start with a small helper function. This function will make a call to a URL and return the value / error wrapped in the appropriate monad object

def make_get_request(url):
    try:
        response = requests.get(url)
        code = response.status_code
        if code == 200:
            return Ok(response.json())
        return Error(f"status {code} from {url}")
    except Exception as e:
        return Error(f"Error connecting to {url}")

Next we write individual functions for each of the api calls. These just call the helper function above with the right url and then map a function to extract the correct field from the response.

def get_ip():
    return (make_get_request("https://api.ipify.org/?format=json")
            .map(lambda data: data["ip"]))

def get_country_from_ipwhois(ip):
    return (make_get_request(f"http://ipwho.is/{ip}")
            .map(lambda data: data["country"]))

def get_country_code_from_ipinfo(ip):
    return (make_get_request(f"https://ipinfo.io/{ip}/geo")
            .map(lambda data: data["country"]))

def get_country_from_country_code(country_code):
    return (make_get_request(f"https://countrycode.dev/api/countries/iso2/{country_code}")
            .map(lambda data: data[0]["country_name"]))

Finally we sequence the calls according to the flowchart

def get_country_from_ip(ip):
    match output := get_country_from_ipwhois(ip):
        case Ok():
            return output
        case Error():
            return (get_country_code_from_ipinfo(ip)
                    .flatmap(get_country_from_country_code))

def get_country():
    match output := get_ip().flatmap(get_country_from_ip):
        case Ok():
            print(f"You are from {output.value}")
        case Error():
            print(output.error)

Run this code and it will print the country. You can block any of the above urls in your firewall to test the various error flows. Blocking ipwho.is will still print the country via the backup flow.

In the usual coding style, we need to handle error after each and every api call. With monads, we can execute the whole flow and check for error at the end to see if the sequence succeeded or failed.