jroachgolf84 commented on code in PR #65314:
URL: https://github.com/apache/airflow/pull/65314#discussion_r3529118450
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -1137,6 +1155,124 @@ def apply_state_filter(query):
else:
tis_full = apply_state_filter(tis_full)
+ if include_dependent_dags:
+ # Recursively find external tasks indicated by ExternalTaskMarker
+ import pendulum
+
+ from airflow.providers.standard.sensors.external_task import
ExternalTaskMarker
+
+ # Build a full-object query for identifying ExternalTaskMarker TIs
in the current set
+ if as_pk_tuple:
+ all_ti_rows = session.execute(tis_pk).all()
+ condition = TaskInstance.filter_for_tis(
+ TaskInstanceKey(**cols._mapping) for cols in all_ti_rows
+ )
+ marker_query = select(TaskInstance).where(condition) if
condition is not None else None
+ else:
+ marker_query = tis_full
+
+ if marker_query is not None:
+ if visited_external_tis is None:
+ visited_external_tis = set()
+
+ external_marker_tis = session.scalars(
+ marker_query.where(TaskInstance.operator ==
ExternalTaskMarker.__name__)
+ )
+
+ for ti in external_marker_tis:
+ ti_key = ti.key.primary
+
+ if ti_key in visited_external_tis:
+ continue
+
+ visited_external_tis.add(ti_key)
+ task: ExternalTaskMarker = cast(
+ "ExternalTaskMarker",
copy.copy(self.get_task(ti.task_id))
+ )
+
+ if max_recursion_depth is None:
+ # Maximum recursion depth is set from the first
ExternalTaskMarker encountered
+ max_recursion_depth = task.recursion_depth
+
+ if recursion_depth + 1 > max_recursion_depth:
+ raise MaxRecursionDepthError(
+ f"Maximum recursion depth {max_recursion_depth}
reached for "
+ f"{ExternalTaskMarker.__name__} {ti.task_id}. "
+ f"Attempted to clear too many tasks or there may
be a cyclic dependency."
+ )
+
+ # Resolve the logical_date that the ExternalTaskMarker
points to
+ dr_logical_date = session.scalar(
+ select(DagRun.logical_date).where(
+ DagRun.dag_id == ti.dag_id, DagRun.run_id ==
ti.run_id
+ )
+ )
+
+ if dr_logical_date is None:
+ continue
+
+ logical_date_str: str = dr_logical_date.isoformat()
+
+ # Check whether a non-default template was used by
comparing the serialized template
+ # field on the task against the default "{{
logical_date.isoformat() }}" pattern
+ default_template = "{{ logical_date.isoformat() }}"
+
+ # If it differs, look up the rendered value from
RenderedTaskInstanceFields
+ if task.logical_date != default_template:
+ from airflow.models.renderedtifields import
RenderedTaskInstanceFields
+
+ rendered =
RenderedTaskInstanceFields.get_templated_fields(ti, session=session)
Review Comment:
Going to create a follow-up issue for this edge case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
