Merge pull request #994 from sul-dlss/t992-vma-email-processing

Refactors VMA email module for clarity and to fix incorrect email
sul-dlss · May 20, 2024 · 913ab61 · 913ab61
2 parents 85f617b + 31c7922
commit 913ab61
Show file tree

Hide file tree

Showing 8 changed files with 259 additions and 377 deletions.
diff --git a/docker-compose.prod.yaml b/docker-compose.prod.yaml
@@ -93,7 +93,7 @@ x-airflow-common:
     AIRFLOW__WEBSERVER__ALLOWED_PAYLOAD_SIZE: 250.0
     AIRFLOW__WEBSERVER__SHOW_TRIGGER_FORM_IF_NO_PARAMS: 'true'
     AWS_DEFAULT_REGION: us-west-2
-    VENDOR_LOADS_TO_EMAIL: ${VENDOR_LOADS_TO_EMAIL:-folio-load-reports@lists.stanford.edu}
+    AIRFLOW_VAR_VENDOR_LOADS_TO_EMAIL: ${VENDOR_LOADS_TO_EMAIL:-folio-load-reports@lists.stanford.edu}
     SQLALCHEMY_SILENCE_UBER_WARNING: 1
     SQLALCHEMY_WARN_20: 1
     # WARNING: Use _PIP_ADDITIONAL_REQUIREMENTS option ONLY for a quick checks

diff --git a/docker-compose.yaml b/docker-compose.yaml
@@ -78,6 +78,7 @@ x-airflow-common:
     AIRFLOW_VAR_MIGRATION_PASSWORD: ${AIRFLOW_VAR_MIGRATION_PASSWORD}
     AIRFLOW_VAR_OKAPI_URL: ${AIRFLOW_VAR_OKAPI_URL}
     AIRFLOW_CONN_VENDOR_LOADS: postgresql+psycopg2://airflow:airflow@postgres/vendor_loads
+    AIRFLOW_VAR_VENDOR_LOADS_TO_EMAIL: ${VENDOR_LOADS_TO_EMAIL:-folio-load-reports@lists.stanford.edu}
     AWS_DEFAULT_REGION: us-west-2
     SQLALCHEMY_SILENCE_UBER_WARNING: 1
     SQLALCHEMY_WARN_20: 1

diff --git a/libsys_airflow/dags/vendor/data_fetch_scheduler.py b/libsys_airflow/dags/vendor/data_fetch_scheduler.py
@@ -10,19 +10,15 @@
 from airflow.providers.postgres.hooks.postgres import PostgresHook
 
 from libsys_airflow.plugins.vendor.models import VendorInterface
-from libsys_airflow.plugins.vendor.emails import email_args
 
 logger = logging.getLogger(__name__)
 
-default_args = dict(
-    {
-        "owner": "folio",
-        "depends_on_past": False,
-        "retries": 0,
-        "retry_delay": timedelta(minutes=5),
-    },
-    **email_args(),
-)
+default_args = {
+    "owner": "folio",
+    "depends_on_past": False,
+    "retries": 0,
+    "retry_delay": timedelta(minutes=5),
+}
 
 with DAG(
     dag_id="data_fetcher_scheduler",

diff --git a/libsys_airflow/dags/vendor/data_fetcher.py b/libsys_airflow/dags/vendor/data_fetcher.py
@@ -11,7 +11,7 @@
 from libsys_airflow.plugins.vendor.download import ftp_download_task
 from libsys_airflow.plugins.vendor.archive import archive_task
 from libsys_airflow.plugins.vendor.paths import download_path
-from libsys_airflow.plugins.vendor.emails import email_args, files_fetched_email_task
+from libsys_airflow.plugins.vendor.emails import files_fetched_email_task
 
 logger = logging.getLogger(__name__)
 
@@ -29,7 +29,6 @@
         "retries": 0,
         "retry_delay": timedelta(minutes=5),
     },
-    **email_args(),
 )
 
 with DAG(
@@ -94,10 +93,4 @@ def setup():
         params["vendor_interface_uuid"],
     )
 
-    files_fetched_email_task(
-        params["vendor_interface_name"],
-        params["vendor_code"],
-        params["vendor_interface_uuid"],
-        downloaded_files,
-        params["environment"],
-    )
+    files_fetched_email_task(downloaded_files, params)
diff --git a/libsys_airflow/dags/vendor/default_data_processor.py b/libsys_airflow/dags/vendor/default_data_processor.py
@@ -13,7 +13,6 @@
     data_import_branch_task,
 )
 from libsys_airflow.plugins.vendor.emails import (
-    email_args,
     file_loaded_email_task,
     file_not_loaded_email_task,
 )
@@ -41,7 +40,6 @@
         "retries": 0,
         "retry_delay": timedelta(minutes=5),
     },
-    **email_args(),
 )
 
 with DAG(
@@ -82,8 +80,8 @@ def setup():
 
         pg_hook = PostgresHook("vendor_loads")
         with Session(pg_hook.get_sqlalchemy_engine()) as session:
-            vendor_interface = VendorInterface.load(
-                params["vendor_interface_uuid"], session
+            vendor_interface = VendorInterface.load_with_vendor(
+                params["vendor_uuid"], params["vendor_interface_uuid"], session
             )
             params["vendor_code"] = vendor_interface.vendor.vendor_code_from_folio
             params["vendor_interface_name"] = vendor_interface.display_name
@@ -154,32 +152,16 @@ def setup():
     )
 
     file_loaded_sensor = file_loaded_sensor_task(
-        params["vendor_interface_uuid"], filename, data_import["job_execution_id"]
+        params["vendor_interface_uuid"],
+        params["filename"],
+        data_import["job_execution_id"],
     )
 
     job_summary = job_summary_task(data_import["job_execution_id"])
 
-    file_loaded_email_task(
-        params["vendor_code"],
-        params["vendor_interface_name"],
-        params["vendor_interface_uuid"],
-        data_import["job_execution_id"],
-        params["download_path"],
-        filename,
-        params["start_time"],
-        processed_params["records_count"],
-        job_summary["srs_stats"],
-        job_summary["instance_stats"],
-        params["environment"],
-    )
+    file_loaded_email_task(processed_params, params)
 
-    file_not_loaded_email = file_not_loaded_email_task(
-        params["vendor_interface_name"],
-        params["vendor_code"],
-        params["vendor_interface_uuid"],
-        filename,
-        params["environment"],
-    )
+    file_not_loaded_email = file_not_loaded_email_task(processed_params, params)
 
     data_import_branch >> data_import >> file_loaded_sensor >> job_summary
     data_import_branch >> file_not_loaded_email
diff --git a/libsys_airflow/dags/vendor/purge_archive.py b/libsys_airflow/dags/vendor/purge_archive.py
@@ -10,7 +10,6 @@
     remove_downloads_task,
     set_status_task,
 )
-from libsys_airflow.plugins.vendor.emails import email_args
 
 logger = logging.getLogger(__name__)
 
@@ -23,7 +22,6 @@
         "retries": 1,
         "retry_delay": timedelta(minutes=5),
     },
-    **email_args(),
 )