Enable Support for Packaged DAGs

deployments/scripts/azure/output_dag_folder.py

@@ -7,12 +7,11 @@ class OutputDag:
         print(f"dag_name: {self.dag_name}")
 
     def create(self):
-        self.copy_dag_folder()
+        self.prepare_zip_folder()
         self.create_workflow_request_body()
 
-    def copy_dag_folder(self):
-        shutil.copytree('./src/dags/', './output_dags/dags')
-        shutil.copytree('./src/plugins/', './output_dags/plugins')
+    def prepare_zip_folder(self):
+        shutil.make_archive('./output_dags/dags/manifest_ingestion_dags', 'zip', './src/osdu_dags')
 
     def create_workflow_request_body(self):
         file_name = "workflow_request_body.json"

src/dags/.airflowignore

-configs
-libs
-providers

src/dags/osdu-ingest-r2.py → src/osdu_dags/osdu-ingest-r2.py

@@ -19,10 +19,10 @@ from datetime import timedelta
 
 import airflow
 from airflow import DAG
-from libs.airflow.backward_compatibility.default_args import update_default_args
-from operators.deprecated.update_status import UpdateStatusOperator
-from operators.process_manifest_r2 import ProcessManifestOperatorR2
-from operators.search_record_id import SearchRecordIdOperator
+from osdu_manifest.libs.airflow.backward_compatibility.default_args import update_default_args
+from osdu_manifest.operators.deprecated.update_status import UpdateStatusOperator
+from osdu_manifest.operators.process_manifest_r2 import ProcessManifestOperatorR2
+from osdu_manifest.operators.search_record_id import SearchRecordIdOperator
 
 default_args = {
     "start_date": airflow.utils.dates.days_ago(0),

src/dags/osdu-ingest-r3.py → src/osdu_dags/osdu-ingest-r3.py

@@ -22,12 +22,12 @@ from airflow import DAG
 from airflow.models import Variable
 from airflow.operators.dummy_operator import DummyOperator
 from airflow.operators.python_operator import BranchPythonOperator
-from libs.airflow.backward_compatibility.default_args import update_default_args
-from operators.ensure_manifest_integrity import EnsureManifestIntegrityOperator
-from operators.process_manifest_r3 import ProcessManifestOperatorR3
-from operators.update_status import UpdateStatusOperator
-from operators.validate_manifest_schema import ValidateManifestSchemaOperator
 from osdu_api.libs.exceptions import NotOSDUSchemaFormatError
+from osdu_manifest.libs.airflow.backward_compatibility.default_args import update_default_args
+from osdu_manifest.operators.ensure_manifest_integrity import EnsureManifestIntegrityOperator
+from osdu_manifest.operators.process_manifest_r3 import ProcessManifestOperatorR3
+from osdu_manifest.operators.update_status import UpdateStatusOperator
+from osdu_manifest.operators.validate_manifest_schema import ValidateManifestSchemaOperator
 
 BATCH_NUMBER = int(Variable.get("core__ingestion__batch_count", "3"))
 PROCESS_SINGLE_MANIFEST_FILE = "process_single_manifest_file_task"

src/plugins/operators/ensure_manifest_integrity.py → src/osdu_dags/osdu_manifest/operators/ensure_manifest_integrity.py

@@ -17,12 +17,12 @@
 import logging
 
 from airflow.models import BaseOperator, Variable
-from libs.airflow.backward_compatibility.airflow_utils import apply_defaults
-from operators.mixins.ReceivingContextMixin import ReceivingContextMixin
 from osdu_api.libs.context import Context
 from osdu_api.libs.refresh_token import AirflowTokenRefresher
 from osdu_api.libs.validation.validate_file_source import FileSourceValidator
 from osdu_api.libs.validation.validate_referential_integrity import ManifestIntegrity
+from osdu_manifest.libs.airflow.backward_compatibility.airflow_utils import apply_defaults
+from osdu_manifest.operators.mixins.ReceivingContextMixin import ReceivingContextMixin
 
 logger = logging.getLogger()