Download the raw, unfiltered traffic data
To produce the traffic data work items for the consumer robot, you need first to get your hands on the raw source data. You create a new method and implement the download logic there:
from robocorp.tasks import task
from RPA.HTTP import HTTP
http = HTTP()
@task
def produce_traffic_data():
"""
Inhuman Insurance, Inc. Artificial Intelligence System automation.
Produces traffic data work items.
"""
print("produce")
download_traffic_data()
@task
def consume_traffic_data():
"""
Inhuman Insurance, Inc. Artificial Intelligence System robot.
Consumes traffic data work items.
"""
print("consume")
def download_traffic_data():
http.download(
url="https://github.com/robocorp/inhuman-insurance-inc/raw/main/RS_198.json",
target_file="output/traffic.json",
overwrite=True,
)
- The
download_traffic_data()method encapsulates the downloading logic nicely. - The RPA.HTTP library provides the download() function for downloading the JSON file.
You run the producer task. The traffic.json file is downloaded and stored inside the output folder. Now you have the raw data available!
You observe that download_traffic_data() does only one other function call.
And we also know from the specifications that we will not need to do this again in our workflow.
So instead of having a function that has such a small responsibility, we are goint to refactor it and merge it with our @task function.
from robocorp.tasks import task
from RPA.HTTP import HTTP
http = HTTP()
@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="output/traffic.json",
overwrite=True,
)
@task
def consume_traffic_data():
"""
Inhuman Insurance, Inc. Artificial Intelligence System robot.
Consumes traffic data work items.
"""
print("consume")