From 0172464ad13004bc3f54e60e7ac1770e6c48eb32 Mon Sep 17 00:00:00 2001 From: dert1129 Date: Wed, 20 May 2026 11:26:50 -0400 Subject: [PATCH] mark packages as failed when the state is upload failed --- data_management/services/dlu_management.py | 7 ++++++- data_management/services/dlu_mongo.py | 4 ++++ data_management/watch_files.py | 16 +++++++++++++++- 3 files changed, 25 insertions(+), 2 deletions(-) diff --git a/data_management/services/dlu_management.py b/data_management/services/dlu_management.py index 1bb0dc0..2195eb5 100644 --- a/data_management/services/dlu_management.py +++ b/data_management/services/dlu_management.py @@ -77,6 +77,11 @@ def get_participant_by_redcap_id(self, redcap_id: str): return self.db.get_data( "SELECT * FROM data_management.redcap_participant WHERE redcap_id = %s",(redcap_id,), ) + + def set_dlu_package_error(self, package_id: str): + return self.db.insert_data( + "UPDATE dlu_package_inventory SET dlu_error = 1 WHERE dlu_package_id = %s", (package_id,) + ) def insert_dlu_package(self, dpi_values: tuple, dmd_values: tuple): @@ -212,7 +217,7 @@ def get_data_manager_data(self): return result def get_package(self, package_id: str) -> dict: - result = self.db.get_data("SELECT * dlu_package_inventory WHERE dlu_package_id = %s", (package_id,)) + result = self.db.get_data("SELECT * FROM dlu_package_inventory WHERE dlu_package_id = %s", (package_id,)) if result: return result[0] else: diff --git a/data_management/services/dlu_mongo.py b/data_management/services/dlu_mongo.py index f1ef6e7..4e56bab 100644 --- a/data_management/services/dlu_mongo.py +++ b/data_management/services/dlu_mongo.py @@ -25,6 +25,7 @@ class DLUMongo: def __init__(self, mongo_connection: MongoConnection): self.package_collection = mongo_connection.packages.with_options(codec_options=CodecOptions(tz_aware=True)) + self.state_collection = mongo_connection.state.with_options(codec_options=CodecOptions(tz_aware=True)) def get_modification_info(self, file_info: dict): modifications = [] @@ -63,6 +64,9 @@ def update_package_files(self, package_id: str, file_info: dict) -> int: result = self.package_collection.update_one({"_id": package_id}, {"$set": {"files": mongo_files, "modifications": final_modifications}}) return result.modified_count + + def find_by_upload_failed(self): + return self.state_collection.find({"state": "UPLOAD_FAILED"}) def find_by_package_type_and_redcap_id(self, package_type: str, subject_id: str): return self.package_collection.find_one({"subjectId": subject_id, "packageType": package_type}) diff --git a/data_management/watch_files.py b/data_management/watch_files.py index 8caafed..a5dcff7 100644 --- a/data_management/watch_files.py +++ b/data_management/watch_files.py @@ -196,7 +196,20 @@ def is_directory_valid(self, directory_info, package_id): self.dlu_management.update_dlu_package(package_id, {"globus_dlu_status": error_msg}) return False return True - + + def mark_packages_with_error(self): + failed_uploads = self.dlu_mongo.find_by_upload_failed() + for failed_upload in failed_uploads: + package_id = failed_upload["packageId"] + package = self.dlu_management.get_package(package_id) + if package is not None: + if package['dlu_error'] == 1: + continue + else: + self.dlu_management.set_dlu_package_error(failed_upload["packageId"]) + logger.info("Marked package " + failed_upload["packageId"] + " as error due to UPLOAD_FAILED in MongoDB") + + if __name__ == "__main__": dlu_watcher = DLUWatcher() @@ -205,4 +218,5 @@ def is_directory_valid(self, directory_info, package_id): dlu_watcher.watch_for_packages() dlu_watcher.watch_for_side_manifest_records() dlu_watcher.fill_in_null_package_ids() + dlu_watcher.mark_packages_with_error() time.sleep(60)