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.

โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ โ”‚ โ”‚ โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜

You rename the file (generated by the template) to to communicate the robot's role. You leave the produce_traffic_data() logic here, and all the functions related to it.

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. """ url="", 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

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: exception_type="APPLICATION", code="TRAFFIC_DATA_POST_FAILED", message=return_json["message"], ) else: exception_type="BUSINESS", code="INVALID_TRAFFIC_DATA", message=item.payload, ) def post_traffic_data_to_sales_system(traffic_data): url = "" response =, 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 Consume data: shell: python -m robocorp.tasks run

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!