Create output work items
Currently, the API payloads are created only in memory - they are not persisted. It's time to save the payloads as work items. You decide to create one work item per API payload:
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 automation.
Consumes traffic data work items.
"""
print("consume")
def load_traffic_data_as_table():
json_data = json.load_json_from_file(TRAFFIC_JSON_FILE_PATH)
return table.create_table(json_data["value"])
def filter_and_sort_traffic_data(data):
max_rate = 5.0
both_genders = "BTSX"
table.filter_table_by_column(data, RATE_KEY, "<", max_rate)
table.filter_table_by_column(data, GENDER_KEY, "==", both_genders)
table.sort_table_by_column(data, YEAR_KEY, False)
return data
def get_latest_data_by_country(data):
data = table.group_table_by_column(data, COUNTRY_KEY)
latest_data_by_country = []
for group in data:
first_row = table.pop_table_row(group)
latest_data_by_country.append(first_row)
return latest_data_by_country
def create_work_item_payloads(traffic_data):
payloads = []
for row in traffic_data:
payload = dict(
country=row[COUNTRY_KEY],
year=row[YEAR_KEY],
rate=row[RATE_KEY],
)
payloads.append(payload)
return payloads
def save_work_item_payloads(payloads):
for payload in payloads:
variables = dict(traffic_data=payload)
workitems.outputs.create(variables)
save_work_item_payloads() is your custom function that takes the list of payloads as an argument and saves each payload as a work item.
The robocorp.workitems library provides a useful function for our current need:
create() creates a new output work item. We store our payload as a dictionary with the given variable name (traffic_data).
After running the producer task, a devdata/work-items-out/run-1/work-items.json file is automatically created. A new run-n folder is created every time you run your robot. Those output work items come in handy later when implementing and testing your consumer robot since they can be used as test input for the consumer.
The work-items.json contains your output work items:
[
{
"payload": {
"traffic_data": {
"country": "VCT",
"year": 2011,
"rate": 3.69293
}
},
"files": {}
},
{
"payload": {
"traffic_data": {
"country": "SWError",
"year": 2019,
"rate": 3.13947
}
},
"files": {}
},...
Each work item has the payload and files properties. In your case, only the payload contains data (the traffic data your robot created). The files property is empty since this robot does not use files.
Your producer robot is now complete!
The producer:
- downloads the raw traffic data.
- transforms the raw data into a business data format.
- saves the business data as work items that can be consumed later.
Well done, you! Time to consume stuff, next.