Hello Guys!
Today, I’m going to discuss a potential use case, where on many occasions, different teams need almost similar kinds of data through API. However, they are not identical. Creating a fresh API/Microservice after following-up with many processes will take significant time.
What if we can create an API in such a way so that we can get the response dynamically without needing to make another one. In this post, we’ll be demonstrating a similar approach.
I’ll be using open-source Covid-API, which will be useful for several posts starting from this one.
You will get plenty of useful data from here.
We’ve chosen the following one for our use case –

Let’s explore the sample data first.
[ { "date":20210207, "state":"AK", "positive":53279.0, "probableCases":null, "negative":null, "pending":null, "totalTestResultsSource":"totalTestsViral", "totalTestResults":1536911.0, "hospitalizedCurrently":44.0, "hospitalizedCumulative":1219.0, "inIcuCurrently":null, "inIcuCumulative":null, "onVentilatorCurrently":11.0, "onVentilatorCumulative":null, "recovered":null, "dataQualityGrade":"A", "lastUpdateEt":"2\/5\/2021 03:59", "dateModified":"2021-02-05T03:59:00Z", "checkTimeEt":"02\/04 22:59", "death":279.0, "hospitalized":1219.0, "dateChecked":"2021-02-05T03:59:00Z", "totalTestsViral":1536911.0, "positiveTestsViral":64404.0, "negativeTestsViral":1470760.0, "positiveCasesViral":null, "deathConfirmed":null, "deathProbable":null, "totalTestEncountersViral":null, "totalTestsPeopleViral":null, "totalTestsAntibody":null, "positiveTestsAntibody":null, "negativeTestsAntibody":null, "totalTestsPeopleAntibody":null, "positiveTestsPeopleAntibody":null, "negativeTestsPeopleAntibody":null, "totalTestsPeopleAntigen":null, "positiveTestsPeopleAntigen":null, "totalTestsAntigen":null, "positiveTestsAntigen":null, "fips":"02", "positiveIncrease":0, "negativeIncrease":0, "total":53279, "totalTestResultsIncrease":0, "posNeg":53279, "deathIncrease":0, "hospitalizedIncrease":0, "hash":"07a5d43f958541e9cdabb5ea34c8fb481835e130", "commercialScore":0, "negativeRegularScore":0, "negativeScore":0, "positiveScore":0, "score":0, "grade":"" } ]
Let’s take two cases. One, where one service might need to access all the elements, there might be another, where some other service requires specific details.
Let’s explore the code base first –
- init.py ( This native Python-based azure-function that will consume streaming data & dynamic API response. )
########################################### #### Written By: SATYAKI DE #### #### Written On: 06-Feb-2021 #### #### Package Flask package needs to #### #### install in order to run this #### #### script. #### #### #### #### Objective: Main Calling scripts. #### #### #### #### However, to meet the functionality#### #### we've enhanced as per our logic. #### ########################################### import logging import json import requests import os import pandas as p import numpy as np import azure.functions as func def main(req: func.HttpRequest) -> func.HttpResponse: logging.info('Dynamic-Covid-Status HTTP trigger function processed a request.') try: # Application Variable url = os.environ['URL'] appType = os.environ['appType'] conType = os.environ['conType'] # API-Configuration payload={} headers = { "Connection": conType, "Content-Type": appType } # Validating input parameters typeSel = req.params.get('typeSel') if not typeSel: try: req_body = req.get_json() except ValueError: pass else: typeSel = req_body.get('typeSel') typeVal = req.params.get('typeVal') if not typeVal: try: req_body = req.get_json() except ValueError: pass else: typeVal = req_body.get('typeVal') # Printing Key-Element Values str1 = 'typeSel: ' + str(typeSel) logging.info(str1) str2 = 'typeVal: ' + str(typeVal) logging.info(str2) # End of API-Inputs # Getting Covid data from the REST-API response = requests.request("GET", url, headers=headers, data=payload) ResJson = response.text if typeSel == '*': if typeVal != '': # Converting it to Json jdata = json.loads(ResJson) df_ret = p.io.json.json_normalize(jdata) df_ret.columns = df_ret.columns.map(lambda x: x.split(".")[-1]) rJson = df_ret.to_json(orient ='records') return func.HttpResponse(rJson, status_code=200) else: x_stat = 'Failed' x_msg = 'Important information is missing for all values!' rJson = { "status": x_stat, "details": x_msg } xval = json.dumps(rJson) return func.HttpResponse(xval, status_code=200) elif typeSel == 'Cols': if typeVal != '': # Converting it to Json jdata = json.loads(ResJson) df_ret = p.io.json.json_normalize(jdata) df_ret.columns = df_ret.columns.map(lambda x: x.split(".")[-1]) # Fetching for the selected columns # Extracting the columns from the list lstHead = [] listX = typeVal.split (",") for i in listX: lstHead.append(str(i).strip()) str3 = 'Main List: ' + str(lstHead) logging.info(str3) slice_df = df_ret[np.intersect1d(df_ret.columns, lstHead)] rJson = slice_df.to_json(orient ='records') return func.HttpResponse(rJson, status_code=200) else: x_stat = 'Failed' x_msg = 'Important information is missing for selected values!' rJson = { "status": x_stat, "details": x_msg } xval = json.dumps(rJson) return func.HttpResponse(xval, status_code=200) else: x_stat = 'Failed' x_msg = 'Important information is missing for typeSel!' rJson = { "status": x_stat, "details": x_msg } xval = json.dumps(rJson) return func.HttpResponse(xval, status_code=200) except Exception as e: x_msg = str(e) x_stat = 'Failed' rJson = { "status": x_stat, "details": x_msg } xval = json.dumps(rJson) return func.HttpResponse(xval, status_code=200)
And, Inside the azure portal it looks like –

Let’s explain the key snippet –
jdata = json.loads(ResJson) df_ret = p.io.json.json_normalize(jdata) df_ret.columns = df_ret.columns.map(lambda x: x.split(".")[-1]) rJson = df_ret.to_json(orient ='records') return func.HttpResponse(rJson, status_code=200)
In the above lines, we’re converting the response & organizing it to a pandas dataframe before converting the response to JSON.
# Fetching for the selected columns # Extracting the columns from the list lstHead = [] listX = typeVal.split (",") for i in listX: lstHead.append(str(i).strip()) str3 = 'Main List: ' + str(lstHead) logging.info(str3) #slice_df = df_ret[df_ret.columns.intersection(lstHead)] slice_df = df_ret[np.intersect1d(df_ret.columns, lstHead)]
For the second case, the above additional logic will play a significant part. Based on the supplied input in the typeVal attribute, this time, the new response will display accordingly.
Let’s see how it looks –

Case 1 (For all the columns):

And, the formatted output is as follows –

Case 2 (For selected columns):


You can find the code in the Github using the following link.
So, finally, we have done it.
I’ll bring some more exciting topic in the coming days from the Python verse.
Till then, Happy Avenging! 😀
Note: All the data & scenario posted here are representational data & scenarios & available over the internet & for educational purpose only.
You must be logged in to post a comment.