Break your robot into multiple files
Now that we have our producer and consumer working, the code file got pretty long.
And actually, if you think about it, the producer does one thing, and the consumer does a different thing.
That means, we could actually split them up.
Sometimes, less (files) is not more (maintainability).
RPA Artificial Intelligence Marketing Demystified!
Maintenance-wise, it might be better to split the robot into logical parts.
Need to update the producer logic? Find the producer file. Consumer logic - consumer file.
Easy peasy! Breaking code down into smaller pieces is a software development best practice.
How do you know how to structure your code beforehand?
Well, you don't! But you can still have a high-level guess and start with that guess.
The structure will evolve while you write more code.
This is expected and totally fine, don't worry.
Code. Refactor. Code. Refactor. Rinse and repeat.
Refactoring fresh and even old existing code is recommended. Learn more in the OpportunisticRefactoring article by Martin Fowler.
Instead of a single long python file, you decide to split the code into two python files.
One file for the producer robot, another for the consumer robot.
┌────────────────┐ ┌────────────────┐
│ producer.py │ │ consumer.py │
└────────────────┘ └────────────────┘
You rename the tasks.py file (generated by the template) to producer.py to communicate the robot's role.
You leave the produce_traffic_data() logic here, and all the functions related to it.
producer.py:
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)
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)
Next, you create a new file consumer.py:
consumer.py
import requests
from robocorp import workitems
from robocorp.tasks import task
@task
def consume_traffic_data():
"""
Inhuman Insurance, Inc. Artificial Intelligence System automation.
Consumes traffic data work items.
"""
for item in workitems.inputs:
traffic_data = item.payload["traffic_data"]
if len(traffic_data["country"]) == 3:
status, return_json = post_traffic_data_to_sales_system(traffic_data)
if status == 200:
item.done()
else:
item.fail(
exception_type="APPLICATION",
code="TRAFFIC_DATA_POST_FAILED",
message=return_json["message"],
)
else:
item.fail(
exception_type="BUSINESS",
code="INVALID_TRAFFIC_DATA",
message=item.payload,
)
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)
return response.status_code, response.json()
Finally, you update robot.yaml to correctly run your tasks:
tasks:
Produce data:
shell: python -m robocorp.tasks run producer.py
Consume data:
shell: python -m robocorp.tasks run consumer.py
We no longer need to specify the task, since each file contains only one @task function.
Congratulations! Your code is now logically split into multiple files!