Repairs Hub bulk upload script - #58
Conversation
| @dataclass | ||
| class CSV_KEYS: | ||
| description_key = 'Description' | ||
| sor_code_key = "SorCode" | ||
| unique_id_key = 'Unique Id' | ||
| prop_ref_key = 'Property Reference' | ||
| priority_key = "Priority" |
There was a problem hiding this comment.
Defined CSV key values
| with RepairsSession() as db_session: | ||
| budget_code = get_budget_code(db_session, corporate_subjective_code, external_cost_code) | ||
| all_priorities = get_sor_priorities(db_session) | ||
| trade = get_trade(db_session, trade_code) | ||
| contractor = get_contractor(db_session, contractor_reference) | ||
| all_sor_codes = get_sor_codes(db_session, extracted_sor_codes) |
There was a problem hiding this comment.
Fetches all data up front. There arent many priorities so we can fetch all. For sor codes, map the requested values to a set, and fetch all matching values. Then we must validate a matching result
| results = csv_to_dict_list(Config.SOURCE_FILE_PATH, is_tsv=False)[:5] | ||
| completed = load_completed_jobs(Config.LOG_FILE_PATH) | ||
|
|
||
| # Filter out completed jobs | ||
| results = [row for row in results if str(row[CSV_KEYS.unique_id_key]) not in completed] | ||
|
|
||
| if not results: | ||
| print("Nothing left to process.") | ||
| return |
There was a problem hiding this comment.
We log completed records to a textfile. Meaning we can easily rerun a failed script without uploading the same job twice.
| # Temporary hardcoded values (this should all be the same for a given bulk upload) | ||
| trade_code = "PL" | ||
| contractor_reference = "RG2" | ||
| corporate_subjective_code="200045" | ||
| external_cost_code="H2555" |
There was a problem hiding this comment.
These values will be the same for a given bulk upload. Im not too sure how best to define them. Adding them to the csv seems redundant.
There was a problem hiding this comment.
This is fine I think - alternatively they can be defined as constants at the module-level at the top of this file under the imports, or they can be arguments for the main function
There was a problem hiding this comment.
Actually since this already uses the Config object, may as well add them there
LBHCallumM
changed the title
|
|
||
| success = create_work_order_via_api(request_body) | ||
|
|
||
| return property_reference, success |
There was a problem hiding this comment.
It would be nice to split this process in 2
- Create all the work order request objects then write them to a json file for manual review
- After review, read from the file and create all the work orders
This workflow is nice for identifying any unexpected issues with the data at a glance
| with progress.Bar("Generating request payloads", max=len(results)) as progress_bar: | ||
| build_errors: list[tuple[str, str]] = [] | ||
|
|
||
| with ThreadPoolExecutor(max_workers=Config.THREAD_POOL_COUNT) as executor: | ||
| futures = { | ||
| executor.submit(build_work_order_payload, row, budget_code, trade, all_sor_codes[row[CsvKeys.sor_code_key]], contractor): row[CsvKeys.unique_id_key] | ||
| for row in results | ||
| } |
There was a problem hiding this comment.
Separated generating payloads from creating the orders
| with open(Config.REQUEST_BODY_FILE_PATH, 'w') as filetowrite: | ||
| request_bodies = [job.payload for job in job_list] | ||
| json.dump(request_bodies, filetowrite, indent=4) | ||
|
|
||
| assert input(f"You can confirm the request bodies at '{Config.REQUEST_BODY_FILE_PATH}'. Press y to continue bulk upload") == "y" | ||
|
|
There was a problem hiding this comment.
Manual confirmation step
| def fetch_one(session: Session, stmt: Select[tuple[T]], label: str) -> T: | ||
| try: | ||
| return session.scalars(stmt).one() | ||
| except NoResultFound: | ||
| raise LookupError(f"No {label} found") from None | ||
| except MultipleResultsFound: | ||
| raise LookupError(f"Multiple {label} matched — expected exactly one") from None |
There was a problem hiding this comment.
I don't think this function is needed - if you just run the script with debug and have it pause on exception you'll get all this info for free. Just use session.scalars(stmt).one() directly
| return False | ||
|
|
||
| def get_asset_by_prop_ref(property_reference: str): | ||
| return get_by_secondary_index(asset_dynamodb_table, "AssetId", "assetId", property_reference) |
There was a problem hiding this comment.
I think get rid of this function (only called in one place anyway) and just use get_by_secondary_index directly - it's clear enough what it's doing
2 participants
Summary of Changes
Requirements:
HACKNEY_JWT_WORK_ORDERThis script will take a list of workOrders from a CSV, and uploaded them via a POST request to repairs API. Property data is fetched directly from the AssetDB. Repairs specific information is also fetched from the DB because it requires a specific format.
The script tries to do all the data-fetching up front for performance reasons. However every workorder will be for a different property, so that request is separate.
The script uses concurrency, allowing multiple jobs to be processed similtaniously. This makes a huge difference because most of the time is spent waiting for an API response.
The script additionally tracks completed uploads within a text file. This allows the script to be rerun after a failure, without reuploading the same job twice. This does require a
unique_id_keyvalue. This could potentially be prop_ref if its unique for every job. But Ive picked a different value for this PR