Add a method to find end date for previous success TI

airflow/models/taskinstance.py

@@ -609,6 +609,12 @@ def get_prev_start_date_success() -> pendulum.DateTime | None:
             return None
         return timezone.coerce_datetime(dagrun.start_date)
 
+    def get_prev_end_date_success() -> pendulum.DateTime | None:
+        dagrun = _get_previous_dagrun_success()
+        if dagrun is None:
+            return None
+        return timezone.coerce_datetime(dagrun.end_date)
+
     @cache
     def get_yesterday_ds() -> str:
         return (logical_date - timedelta(1)).strftime("%Y-%m-%d")
@@ -729,6 +735,7 @@ def get_triggering_events() -> dict[str, list[DatasetEvent]]:
             session=session,
         ),
         "prev_start_date_success": get_prev_start_date_success(),
+        "prev_end_date_success": get_prev_end_date_success(),
         "run_id": task_instance.run_id,
         "task": task,
         "task_instance": task_instance,

airflow/operators/python.py

@@ -338,6 +338,7 @@ class _BasePythonVirtualenvOperator(PythonOperator, metaclass=ABCMeta):
         "prev_execution_date",
         "prev_execution_date_success",
         "prev_start_date_success",
+        "prev_end_date_success",
     }
     AIRFLOW_SERIALIZABLE_CONTEXT_KEYS = {
         "macros",

airflow/utils/context.py

@@ -71,6 +71,7 @@
     "prev_execution_date",
     "prev_execution_date_success",
     "prev_start_date_success",
+    "prev_end_date_success",
     "run_id",
     "task",
     "task_instance",

airflow/utils/context.pyi

@@ -83,6 +83,7 @@ class Context(TypedDict, total=False):
     prev_execution_date: DateTime | None
     prev_execution_date_success: DateTime | None
     prev_start_date_success: DateTime | None
+    prev_end_date_success: DateTime | None
     reason: str | None
     run_id: str
     task: BaseOperator

tests/operators/test_python.py

@@ -1073,6 +1073,7 @@ def f(
             prev_execution_date,
             prev_execution_date_success,
             prev_start_date_success,
+            prev_end_date_success,
             # airflow-specific
             macros,
             conf,
@@ -1113,6 +1114,7 @@ def f(
             prev_execution_date,
             prev_execution_date_success,
             prev_start_date_success,
+            prev_end_date_success,
             # other
             **context,
         ):