Allow overriding the date from the DagRun conf (#880)

openverse_catalog/dags/providers/provider_api_scripts/provider_data_ingester.py

@@ -2,6 +2,7 @@
 import logging
 import traceback
 from abc import ABC, abstractmethod
+from datetime import datetime
 
 from airflow.exceptions import AirflowException
 from airflow.models import Variable
@@ -129,6 +130,12 @@ def __init__(self, conf: dict = None, dag_id: str = None, date: str = None):
         # dag_run configuration options
         conf = conf or {}
 
+        # Allow overriding the date with a %Y-%m-%d string from the dagrun conf.
+        date_override = conf.get("date")
+        if date_override and datetime.strptime(date_override, "%Y-%m-%d"):
+            logger.info(f"Using date {date_override} from dagrun conf.")
+            self.date = date_override
+
         # Used to skip over errors and continue ingestion. When enabled, errors
         # are not reported until ingestion has completed.
         self.skip_ingestion_errors = conf.get("skip_ingestion_errors", False)

tests/dags/providers/provider_api_scripts/test_provider_data_ingester.py

@@ -88,6 +88,30 @@ def test_batch_limit_is_capped_to_ingestion_limit():
         assert ingester.limit == 20
 
 
+@pytest.mark.parametrize(
+    "date, date_override, expected_date",
+    [
+        # No override
+        ("2022-01-01", None, "2022-01-01"),
+        # Simple override
+        ("2022-01-01", "2022-12-12", "2022-12-12"),
+        # Incorrect date format throws error
+        pytest.param(
+            "2022-01-01",
+            "12/12/22",
+            None,
+            marks=pytest.mark.raises(exception=ValueError),
+        ),
+    ],
+)
+def test_date_override(date, date_override, expected_date):
+    ingester = MockProviderDataIngester(
+        conf={"date": date_override},  # DagRun conf object
+        date=date,
+    )
+    assert ingester.date == expected_date
+
+
 def test_get_batch_data():
     response_json = _get_resource_json("complete_response.json")
     batch = ingester.get_batch_data(response_json)