Refactor to use dag_run_url Function in utils

libsys_airflow/plugins/data_exports/email.py

@@ -1,6 +1,5 @@
 import logging
 import pathlib
-import urllib
 
 from jinja2 import Template
 
@@ -9,7 +8,7 @@
 from airflow.models import Variable
 from libsys_airflow.plugins.shared.utils import send_email_with_server_name
 
-from libsys_airflow.plugins.shared.utils import is_production
+from libsys_airflow.plugins.shared.utils import is_production, dag_run_url
 
 logger = logging.getLogger(__name__)
 
@@ -68,14 +67,6 @@ def _oclc_report_html(report: str, library: str):
     return f"""{report_type} link: <a href="{report_url}">{report_path.name}</a>"""
 
 
-def dag_run_url(dag_run) -> str:
-    airflow_url = conf.get('webserver', 'base_url')
-    if not airflow_url.endswith("/"):
-        airflow_url = f"{airflow_url}/"
-    params = urllib.parse.urlencode({"dag_run_id": dag_run.run_id})
-    return f"{airflow_url}dags/{dag_run.id}/grid?{params}"
-
-
 def generate_holdings_errors_emails(error_reports: dict):
     """
     Generates emails for holdings set errors for cohort libraries
@@ -227,9 +218,10 @@ def failed_transmission_email(files: list, **kwargs):
     Sends to libsys devs to troubleshoot
     """
     dag_run = kwargs["dag_run"]
-    dag_id = dag_run.id
     dag_run_id = dag_run.run_id
-    run_url = dag_run_url(dag_run)
+    dag_id = dag_run.dag.dag_id
+
+    run_url = dag_run_url(dag_run=dag_run)
     params = kwargs.get("params", {})
     full_dump_vendor = params.get("vendor", {})
     if len(files) == 0:

libsys_airflow/plugins/data_exports/oclc_reports.py

@@ -7,7 +7,7 @@
 from airflow.models import Variable
 from jinja2 import DictLoader, Environment
 
-from libsys_airflow.plugins.data_exports.email import dag_run_url
+from libsys_airflow.plugins.shared.utils import dag_run_url
 
 logger = logging.getLogger(__name__)
 
@@ -276,7 +276,7 @@ def _generate_multiple_oclc_numbers_report(**kwargs) -> dict:
     reports: dict = {}
 
     kwargs["date"] = date.strftime("%d %B %Y")
-    kwargs["dag_run_url"] = dag_run_url(kwargs["dag_run"])
+    kwargs["dag_run_url"] = dag_run_url(dag_run=kwargs["dag_run"])
 
     for library, errors in library_instances.items():
         kwargs["failures"] = errors
@@ -335,7 +335,7 @@ def _reports_by_library(**kwargs) -> dict:
         kwargs["library"] = library
         kwargs["failures"] = filtered_failures
         kwargs["date"] = date.strftime("%d %B %Y")
-        kwargs["dag_run_url"] = dag_run_url(kwargs["dag_run"])
+        kwargs["dag_run_url"] = dag_run_url(dag_run=kwargs["dag_run"])
         reports[library] = report_template.render(**kwargs)
 
     return reports

libsys_airflow/plugins/digital_bookplates/dag_979_sensor.py

@@ -3,6 +3,8 @@
 from airflow.models import DagRun
 from airflow.sensors.base_sensor_operator import BaseSensorOperator
 
+from libsys_airflow.plugins.shared.utils import dag_run_url
+
 logger = logging.getLogger(__name__)
 
 
@@ -23,6 +25,7 @@ def poke(self, context) -> bool:
             dag_run = dag_runs[0]
             state = dag_run.get_state()
             self.dag_runs[dag_run_id]['state'] = state
+            self.dag_runs[dag_run_id]['url'] = dag_run_url(dag_run=dag_run)
             if state in ['success', 'failed']:
                 instances = []
                 for instance, bookplates in dag_run.conf[

libsys_airflow/plugins/digital_bookplates/email.py

@@ -2,7 +2,6 @@
 
 from jinja2 import Template
 
-from airflow.configuration import conf
 from airflow.decorators import task
 from airflow.models import Variable
 
@@ -97,19 +96,14 @@ def _new_updated_bookplates_email_body(new: list, updated: list):
 
 
 def _summary_add_979_email(dag_runs: list, folio_url: str) -> str:
-    airflow_url = conf.get('webserver', 'base_url')  # type: ignore
     if len(dag_runs) < 1:
         return ""
-
-    if not airflow_url.endswith("/"):
-        airflow_url = f"{airflow_url}/"
-    dag_url = f"{airflow_url}dags/digital_bookplate_979/grid?dag_run_id="
     return Template(
         """
         <h2>Results from adding 979 fields Workflows</h2>
         <ol>
         {% for dag_run_id, result in dag_runs.items() %}
-        <li>DAG Run <a href="{{ dag_url }}{{ dag_run_id|urlencode }}">{{ dag_run_id }}</a> {{ result.state }}<br>
+        <li>DAG Run <a href="{{ result.url }}">{{ dag_run_id }}</a> {{ result.state }}<br>
            Instances:
            <ul>
            {% for instance in result.instances %}
@@ -134,7 +128,7 @@ def _summary_add_979_email(dag_runs: list, folio_url: str) -> str:
         {% endfor %}
         </ol>
     """
-    ).render(dag_runs=dag_runs, folio_url=folio_url, dag_url=dag_url)
+    ).render(dag_runs=dag_runs, folio_url=folio_url)
 
 
 def _to_addresses():
@@ -232,6 +226,7 @@ def summary_add_979_dag_runs(**kwargs):
     to_emails = _to_addresses()
     if additional_email:
         to_emails.append(additional_email)
+
     html_content = _summary_add_979_email(dag_runs, folio_url)
 
     if len(html_content.strip()) > 0:

libsys_airflow/plugins/shared/utils.py

@@ -3,8 +3,11 @@
 import logging
 import pymarc
 import re
+import urllib
 
 from typing import Union
+
+from airflow.configuration import conf
 from airflow.models import Variable
 from airflow.utils.email import send_email
 
@@ -13,6 +16,19 @@
 logger = logging.getLogger(__name__)
 
 
+def dag_run_url(**kwargs) -> str:
+    dag_run = kwargs["dag_run"]
+    airflow_url = kwargs.get("airflow_url")
+
+    if not airflow_url:
+        airflow_url = conf.get('webserver', 'base_url')
+        if not airflow_url.endswith("/"):
+            airflow_url = f"{airflow_url}/"
+
+    params = urllib.parse.urlencode({"dag_run_id": dag_run.run_id})
+    return f"{airflow_url}dags/{dag_run.dag.dag_id}/grid?{params}"
+
+
 def is_production():
     return Variable.get("OKAPI_URL").find("prod") > 0
 

tests/data_exports/test_data_exports_emails.py

@@ -49,7 +49,8 @@ def mock_get(key, *args):
 def mock_dag_run(mocker):
     dag_run = mocker.stub(name="dag_run")
     dag_run.run_id = "manual_2022-03-05"
-    dag_run.id = "send_vendor_records"
+    dag_run.dag = mocker.stub(name="dag")
+    dag_run.dag.dag_id = "send_vendor_records"
 
     return dag_run
 
@@ -187,7 +188,7 @@ def test_failed_transmission_email(mocker, mock_dag_run, mock_folio_variables, c
 def test_failed_full_dump_transmission_email(
     mocker, mock_dag_run, mock_folio_variables
 ):
-    mock_dag_run.id = "send_all_records"
+    mock_dag_run.dag.dag_id = "send_all_records"
 
     mock_send_email = mocker.patch(
         "libsys_airflow.plugins.data_exports.email.send_email_with_server_name"

tests/data_exports/test_oclc_reports.py

@@ -17,10 +17,12 @@
 
 @pytest.fixture
 def mock_dag_run(mocker):
-    mock_dag = mocker.MagicMock()
-    mock_dag.id = "send_oclc_records"
-    mock_dag.run_id = "scheduled__2024-07-29T19:00:00:00:00"
-    return mock_dag
+    dag_run = mocker.stub(name="dag_run")
+    dag_run.run_id = "scheduled__2024-07-29T19:00:00:00:00"
+    dag_run.dag = mocker.stub(name="dag")
+    dag_run.dag.dag_id = "send_oclc_records"
+
+    return dag_run
 
 
 @pytest.fixture

tests/digital_bookplates/test_bookplates_emails.py

@@ -82,6 +82,7 @@ def mock_DAG979Sensor():
     return {
         "manual__2024-10-20T02:00:00+00:00": {
             "state": "success",
+            "url": "https://sul-libsys-airflow.stanford.edu/dags/digital_bookplate_979/grid?dag_run_id=manual__2024-10-20T02:00:00+00:00",
             "instances": [
                 {
                     "uuid": "fddf7e4c-161e-4ae8-baad-058288f63e17",
@@ -314,11 +315,6 @@ def test_missing_fields_email_prod(mocker, mock_folio_variables):
 def test_summary_add_979_dag_runs(mocker, mock_DAG979Sensor, mock_folio_variables):
     mock_send_email = mocker.patch("libsys_airflow.plugins.shared.utils.send_email")
 
-    mocker.patch(
-        "libsys_airflow.plugins.digital_bookplates.email.conf.get",
-        return_value="https://sul-libsys-airflow.stanford.edu",
-    )
-
     summary_add_979_dag_runs.function(
         dag_runs=mock_DAG979Sensor, email="[email protected]"
     )
@@ -376,11 +372,6 @@ def test_summary_add_979_dag_runs_prod(mocker, mock_DAG979Sensor, mock_folio_var
 def test_no_summary_add_979_email(mocker, mock_folio_variables):
     mock_send_email = mocker.patch("libsys_airflow.plugins.shared.utils.send_email")
 
-    mocker.patch(
-        "libsys_airflow.plugins.digital_bookplates.email.conf.get",
-        return_value="https://sul-libsys-airflow.stanford.edu",
-    )
-
     summary_add_979_dag_runs.function(dag_runs={}, email="[email protected]")
 
     assert mock_send_email.called is False

tests/digital_bookplates/test_dag_979_sensor.py

@@ -10,6 +10,8 @@ def mock_get_state():
 
     mock_dag_run = MagicMock()
     mock_dag_run.get_state = mock_get_state
+    mock_dag_run.run_id = "manual__2024-10-17"
+    mock_dag_run.dag.dag_id = "digital_bookplate_979"
     mock_dag_run.conf = {
         "druids_for_instance_id": {"d55f7f1b-9512-452c-98ff-5e2be9dcdb16": {}}
     }
@@ -35,3 +37,6 @@ def test_dag_979_sensor(mocker):
     )
     result = sensor.poke(context={})
     assert result is True
+    assert sensor.dag_runs['manual__2024-10-17']["url"].endswith(
+        "digital_bookplate_979/grid?dag_run_id=manual__2024-10-17"
+    )