The import for the prefect state changes, updated all uses.

Author: Fjodor van Rijsselberg

scripts/configuration/odissei_settings.toml

@@ -72,7 +72,7 @@ CBS_TEMPLATE_FILE_PATH="/app/resources/templates/cbs_dataverse_template.json"
 CBS_MAPPING_FILE_PATH="/app/resources/mappings/cbs-mapping.json"
 
 LISS_TEMPLATE_FILE_PATH="/app/resources/templates/liss_dataverse_template.json"
-LISS_MAPPING_FILE_PATH="/resources/mappings/liss-mapping.json"
+LISS_MAPPING_FILE_PATH="/app/resources/mappings/liss-mapping.json"
 
 CID_TEMPLATE_FILE_PATH="/app/resources/templates/cid_dataverse_template.json"
 CID_MAPPING_FILE_PATH="/app/resources/mappings/cid-mapping.json"

scripts/flows/dataset_workflows/cbs_ingestion.py

@@ -1,7 +1,7 @@
 import jmespath
 
 from prefect import flow
-from prefect.server.schemas.states import Completed, Failed
+from prefect.states import Completed, Failed
 from queries import DIST_DATE_QUERY, CBS_ID_QUERY
 from tasks.base_tasks import xml2json, dataverse_mapper, \
     dataverse_import, update_publication_date, add_workflow_versioning_url, \

scripts/flows/dataset_workflows/cid_ingestion.py

@@ -2,7 +2,7 @@
 import jmespath
 
 from prefect import flow, get_run_logger
-from prefect.server.schemas.states import Completed, Failed
+from prefect.states import Completed, Failed
 from queries import DIST_DATE_QUERY
 from tasks.base_tasks import dataverse_mapper, \
     dataverse_import, update_publication_date, add_workflow_versioning_url, \

scripts/flows/dataset_workflows/dataverse_deletion.py

@@ -1,5 +1,5 @@
 from prefect import flow
-from prefect.server.schemas.states import Failed, Completed
+from prefect.states import Failed, Completed
 
 from tasks.base_tasks import dataverse_dataset_check_status, \
     delete_dataset

scripts/flows/dataset_workflows/dataverse_ingestion.py

@@ -1,5 +1,5 @@
 from prefect import flow
-from prefect.server.schemas.states import Failed, Completed
+from prefect.states import Failed, Completed
 
 from tasks.base_tasks import dataverse_metadata_fetcher, dataverse_import, \
     update_publication_date, add_workflow_versioning_url, refine_metadata, \

scripts/flows/dataset_workflows/liss_ingestion.py

@@ -2,7 +2,7 @@
 import jmespath
 
 from prefect import flow
-from prefect.server.schemas.states import Completed, Failed
+from prefect.states import Completed, Failed
 from queries import DIST_DATE_QUERY
 from tasks.base_tasks import dataverse_mapper, \
     dataverse_import, update_publication_date, get_doi_from_dv_json, \

scripts/flows/entry_workflows/main_dataverse_deletion.py

@@ -56,4 +56,4 @@ def dataverse_deletion_pipeline(settings_dict_name: str,
         settings_dict,
         minio_client,
         "identifiers-deleted.json",
-    )
+    )

scripts/utils.py

@@ -113,7 +113,7 @@ def identifier_list_workflow_executor(
         settings_dict,
         s3_client,
         object_name,
-        version = None,
+        version=None,
 ):
     """
     Grabs a file called identifiers.json from the bucket
@@ -122,17 +122,20 @@ def identifier_list_workflow_executor(
     If not return FAILED, if it does execute the data_provider_workflow
     for every pid in the list.
 
-    :param data_provider_workflow: A function representing the data provider workflow.
+    :param object_name: Can be identifiers.json or identifiers-deleted.json
+    :param data_provider_workflow: A function representing the workflow.
     :param version: The version of the workflow to be executed.
-    :param settings_dict: A dictionary containing the settings including the BUCKET_NAME.
-    :param s3_client: An object representing the Boto3 S3 client to access the bucket.
-    :return: 'SUCCESS' if the workflow is executed successfully, 'FAILED' otherwise.
+    :param settings_dict: The settings including the BUCKET_NAME.
+    :param s3_client: An object representing the Boto3 S3 client.
+    :return: 'SUCCESS' if the workflow succeeds, 'FAILED' otherwise.
     """
     bucket_name = settings_dict.BUCKET_NAME
-    identifiers_dict = retrieve_identifiers_from_bucket(s3_client, bucket_name, object_name)
+    identifiers_dict = retrieve_identifiers_from_bucket(s3_client, bucket_name,
+                                                        object_name)
     for pid in identifiers_dict['pids']:
         if version:
-            data_provider_workflow(pid, settings_dict, version, return_state=True)
+            data_provider_workflow(pid, version, settings_dict,
+                                   return_state=True)
         else:
             data_provider_workflow(pid, settings_dict, return_state=True)
 
@@ -244,6 +247,7 @@ def failed_dataverse_ingestion_hook(flow, flow_run, state):
 
     update_identifiers_json(bucket_name, "identifiers.json", pid)
 
+
 def failed_dataverse_deletion_hook(flow, flow_run, state):
     logger = get_run_logger()
     settings_dict = flow_run.parameters["settings_dict"]
@@ -261,7 +265,8 @@ def failed_dataverse_deletion_hook(flow, flow_run, state):
 def update_identifiers_json(bucket_name, object_name, failed_pid):
     s3_client = create_s3_client()
     create_identifiers_json(s3_client, bucket_name, object_name)
-    identifiers_dict = retrieve_identifiers_from_bucket(s3_client, bucket_name, object_name)
+    identifiers_dict = retrieve_identifiers_from_bucket(s3_client, bucket_name,
+                                                        object_name)
     identifiers_dict['pids'].append(failed_pid)
 
     updated_data = json.dumps(identifiers_dict).encode('utf-8')