1446 dag run failed 979s

libsys_airflow/dags/digital_bookplates/retry_failed_979s.py

@@ -0,0 +1,43 @@
+from datetime import datetime
+
+from airflow.decorators import dag
+from airflow.operators.empty import EmptyOperator
+from airflow.timetables.interval import CronDataIntervalTimetable
+
+from libsys_airflow.plugins.digital_bookplates.dag_979_retries import (
+    failed_979_dags,
+    run_failed_979_dags,
+)
+
+
+default_args = {
+    "owner": "libsys",
+    "depends_on_past": False,
+    "email_on_failure": True,
+    "email_on_retry": False,
+    "retries": 0,
+}
+
+
+@dag(
+    default_args=default_args,
+    start_date=datetime(2024, 10, 15),
+    schedule=CronDataIntervalTimetable(
+        cron="45 20 7 * *", timezone="America/Los_Angeles"
+    ),
+    catchup=False,
+    tags=["digital bookplates"],
+)
+def retry_failed_979s():
+    start = EmptyOperator(task_id="start")
+
+    end = EmptyOperator(task_id="end")
+
+    find_failed_979_dags = failed_979_dags()
+
+    rerun_failed_979_dags = run_failed_979_dags(dags=find_failed_979_dags)
+
+    start >> find_failed_979_dags >> rerun_failed_979_dags >> end
+
+
+retry_failed_979s()

libsys_airflow/plugins/digital_bookplates/apps/digital_bookplates_batch_upload_view.py

@@ -12,7 +12,7 @@
 
 from libsys_airflow.plugins.digital_bookplates.bookplates import (
     launch_digital_bookplate_979_dag,
-    launch_poll_for_979_dags,
+    launch_poll_for_979_dags_email,
 )
 from libsys_airflow.plugins.digital_bookplates.models import DigitalBookplate
 
@@ -90,7 +90,7 @@ def trigger_add_979_dags(self):
                 raw_upload_instances_file.filename,
                 upload_instances_df,
             )
-            launch_poll_for_979_dags(dag_runs=dag_runs, email=email)
+            launch_poll_for_979_dags_email(dag_runs=dag_runs, email=email)
             flash(
                 f"Triggered {len(dag_runs)} DAG run(s) for {raw_upload_instances_file.filename}"
             )

libsys_airflow/plugins/digital_bookplates/bookplates.py

@@ -73,7 +73,7 @@ def launch_digital_bookplate_979_dag(**kwargs) -> str:
     return dag_run_id
 
 
-def launch_poll_for_979_dags(**kwargs):
+def launch_poll_for_979_dags_email(**kwargs):
     """
     Triggers poll_for_digital_bookplate_979s_email DAG with kwargs
     """
@@ -302,4 +302,4 @@ def trigger_digital_bookplate_979_task(**kwargs):
 @task
 def trigger_poll_for_979s_task(**kwargs):
     dag_run_ids = kwargs["dag_runs"]
-    launch_poll_for_979_dags(dag_runs=dag_run_ids)
+    launch_poll_for_979_dags_email(dag_runs=dag_run_ids)

libsys_airflow/plugins/digital_bookplates/dag_979_retries.py

@@ -0,0 +1,52 @@
+import logging
+
+from airflow.decorators import task
+from airflow.models import DagBag, DagRun, Variable
+from airflow.utils.state import DagRunState
+
+from libsys_airflow.plugins.digital_bookplates.bookplates import (
+    launch_poll_for_979_dags_email,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@task
+def failed_979_dags() -> dict:
+    """
+    Find all of the failed digital_bookplate_979 DAG runs
+    """
+    dag_runs = DagRun.find(
+        state=DagRunState.FAILED,
+        dag_id="digital_bookplate_979",
+    )
+    db_979_dags: dict = {"digital_bookplate_979s": []}
+    for dag_run in dag_runs:
+        logger.info(f"Found: {dag_run.run_id}")
+        db_979_dags["digital_bookplate_979s"].append(dag_run.run_id)
+
+    return db_979_dags
+
+
+@task
+def run_failed_979_dags(**kwargs):
+    """
+    Re-run the failed digital_bookplate_979 DAGs and launch the email poll
+    """
+    params = kwargs.get("dag_runs", {})
+    devs_email_addr = Variable.get("EMAIL_DEVS")
+
+    dag_runs = params.get("digital_bookplate_979s", {})
+    logger.info(f"Clearing failed 979 DAG runs: {dag_runs}")
+
+    dagbag = DagBag("/opt/airflow/dags")
+    dag = dagbag.get_dag("digital_bookplate_979")
+    dag.clear_dags(
+        dags=[dag],
+        only_failed=True,
+        dry_run=False,
+    )
+
+    launch_poll_for_979_dags_email(dag_runs=dag_runs, email=devs_email_addr)
+
+    return None

tests/digital_bookplates/test_bookplates.py

@@ -11,7 +11,7 @@
     bookplate_funds_polines,
     instances_from_po_lines,
     launch_digital_bookplate_979_dag,
-    launch_poll_for_979_dags,
+    launch_poll_for_979_dags_email,
     trigger_digital_bookplate_979_task,
     _new_bookplates,
 )
@@ -406,7 +406,7 @@ def test_launch_poll_for_979_dags(mocker, mock_dag_bag, caplog):
         return_value=mock_dag_bag,
     )
 
-    launch_poll_for_979_dags(dag_runs=['manual__2024-10-24:00:00:00'])
+    launch_poll_for_979_dags_email(dag_runs=['manual__2024-10-24:00:00:00'])
 
     assert dag_bag.called
     assert "Triggers polling DAG for 979 DAG runs" in caplog.text

tests/digital_bookplates/test_retry_failed_979s.py

@@ -0,0 +1,128 @@
+import pytest
+
+from datetime import datetime
+from unittest.mock import MagicMock
+
+from airflow.models import Variable
+
+from libsys_airflow.plugins.digital_bookplates.dag_979_retries import (
+    failed_979_dags,
+    run_failed_979_dags,
+)
+
+
+def month():
+    return datetime.now().month
+
+
+def uuids():
+    return [
+        "47bf71da-5ca4-496d-a34e-fbc121cd3f1b",
+        "ddb75fc0-1018-4508-bb17-18a1eea57ccf",
+        "399c38dd-51f2-4d96-82b1-82a0e5da7de0",
+        "a483d67a-aae5-48d6-a9cc-e9a061010560",
+        "d062b41a-6809-4f1b-830f-48a3f95df98e",
+    ]
+
+
+def mock_dag_runs():
+    mock_dag_runs = []
+
+    def mock_get_state():
+        return 'failed'
+
+    def mock_get_task_instance(*args):
+        task_instance = MagicMock()
+        task_instance.xcom_pull = mock_xcom_pull
+        return task_instance
+
+    for id, idx in enumerate(uuids()):
+        mock_dag_run = MagicMock()
+        mock_dag_run.get_state = mock_get_state
+        mock_dag_run.run_id = f"scheduled__2024-{month()}-{idx}"
+        mock_dag_run.dag.dag_id = "digital_bookplate_979"
+        mock_dag_run.conf = {"druids_for_instance_id": {id: {}}}
+        mock_dag_run.get_task_instance = mock_get_task_instance
+        mock_dag_runs.append(mock_dag_run)
+
+    return mock_dag_runs
+
+
+def mock_xcom_pull(*args, **kwargs):
+    return {
+        uuids()[-1]: [
+            {
+                'fund_name': 'KLEINH',
+                'druid': 'vy482pt7540',
+                'image_filename': 'vy482pt7540_00_0001.jp2',
+                'title': 'The Herbert A. Klein Book Fund',
+            }
+        ]
+    }
+
+
+@pytest.fixture
+def mock_variable(monkeypatch):
+    def mock_get(key, *args):
+        return "[email protected]"
+
+    monkeypatch.setattr(Variable, "get", mock_get)
+
+
+@pytest.fixture
+def mock_dag_bag(mocker):
+    def mock_get_dag(dag_id: str):
+        return mocker.MagicMock()
+
+    dag_bag = mocker.MagicMock()
+    dag_bag.get_dag = mock_get_dag
+    return dag_bag
+
+
+@pytest.fixture
+def mock_dag(mocker):
+    def mock_clear_dags(dags: list):
+        return mocker.MagicMock()
+
+    dag = mocker.MagicMock()
+    dag.clear_dags = mock_clear_dags
+    return dag
+
+
+def test_find_failed_979_dags(mocker, caplog):
+    mocker.patch(
+        "libsys_airflow.plugins.digital_bookplates.dag_979_retries.DagRun.find",
+        return_value=mock_dag_runs(),
+    )
+
+    failed_dags = failed_979_dags.function()
+    assert len(failed_dags["digital_bookplate_979s"]) == 5
+    assert f"Found: scheduled__2024-{month()}-4" in caplog.text
+
+
+def test_run_failed_979_dags(mocker, mock_variable, mock_dag_bag, mock_dag, caplog):
+    mocker.patch(
+        "libsys_airflow.plugins.digital_bookplates.dag_979_retries.DagRun.find",
+        return_value=mock_dag_runs(),
+    )
+    mocker.patch(
+        "libsys_airflow.plugins.digital_bookplates.dag_979_retries.DagRun.get_dag",
+        return_value=mock_dag,
+    )
+    dag = mocker.patch(
+        "libsys_airflow.plugins.digital_bookplates.bookplates.DagBag",
+        return_value=mock_dag,
+    )
+    dag_bag = mocker.patch(
+        "libsys_airflow.plugins.digital_bookplates.dag_979_retries.DagBag",
+        return_value=mock_dag_bag,
+    )
+
+    dag_runs = {"digital_bookplate_979s": mock_dag_runs()}
+    assert len(dag_runs) > 0
+
+    run_failed_979_dags.function(dag_runs=dag_runs)
+
+    assert dag_bag.called
+    assert dag.called
+    assert "Clearing failed 979 DAG runs" in caplog.text