First attempt at sales system API integration
Time to start the conversation with the sales system API. As an icebreaker, you implement a POST request:
import requests
from robocorp import workitems
from robocorp.tasks import task
from RPA.HTTP import HTTP
from RPA.JSON import JSON
from RPA.Tables import Tables
http = HTTP()
json = JSON()
table = Tables()
TRAFFIC_JSON_FILE_PATH = "output/traffic.json"
# JSON data keys
COUNTRY_KEY = "SpatialDim"
YEAR_KEY = "TimeDim"
RATE_KEY = "NumericValue"
GENDER_KEY = "Dim1"
@task
def produce_traffic_data():
"""
Inhuman Insurance, Inc. Artificial Intelligence System automation.
Produces traffic data work items.
"""
http.download(
url="https://github.com/robocorp/inhuman-insurance-inc/raw/main/RS_198.json",
target_file=TRAFFIC_JSON_FILE_PATH,
overwrite=True,
)
traffic_data = load_traffic_data_as_table()
filtered_data = filter_and_sort_traffic_data(traffic_data)
filtered_data = get_latest_data_by_country(filtered_data)
payloads = create_work_item_payloads(filtered_data)
save_work_item_payloads(payloads)
@task
def consume_traffic_data():
"""
Inhuman Insurance, Inc. Artificial Intelligence System robot.
Consumes traffic data work items.
"""
process_traffic_data()
def process_traffic_data():
for item in workitems.inputs:
traffic_data = item.payload["traffic_data"]
valid = validate_traffic_data(traffic_data)
if valid:
post_traffic_data_to_sales_system(traffic_data)
def validate_traffic_data(traffic_data):
return len(traffic_data["country"]) == 3
def post_traffic_data_to_sales_system(traffic_data):
url = "https://robocorp.com/inhuman-insurance-inc/sales-system-api"
response = requests.post(url, json=traffic_data)
- The
if condition checks if the data is valid. If yes, the post_traffic_data_to_sales_system() function is called. If not, nothing is done.
We are going to explore a bit what else is out there, besides the libraries offered by Robocorp.
Python exists in a big and wide world and there are a lot of libraries waiting in the wild to be used.
For the post calls we are going to use the requests library.
The post() posts the JSON payload to the sales system API endpoint URL.
When you run the consumer task, it will pass succesfully.
Now, we know from the specifications that some of our calls should fail, so this is a bit suspect.
Let us see the calls actually fail from time to time:
def post_traffic_data_to_sales_system(data):
url = "https://robocorp.com/inhuman-insurance-inc/sales-system-api"
response = requests.post(url, json=data)
response.raise_for_status()
Adding response.raise_for_status() will show us that actually not everything is passing.
When you run the consumer task, eventually, one of the POST requests will fail with an exception:
raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 500 Server Error: Internal Server Error for url: https://robocorp.com/inhuman-insurance-inc/sales-system-api
We are not going to keep this, we want our code to pass, so we will remove the response.raise_for_status() line.
We used it just for debugging purposes, to look under the hood of what is actually happening.
With this, your robot gives up immediately after hitting an exception.
We are going to handle the responses in the following steps.