jason810496 opened a new pull request, #74034: URL: https://github.com/apache/airflow/pull/74034
Replaces #74004, which GitHub closed as merged into a stack branch when the stack was reordered. Stack (bottom to top): #73841, **#74004**, #73842, #73843, #73844, #73845, #73846, #73847 - related: #71929 - Part of the native Dag e2e stack. Builds on #73841 ([layer diff](https://github.com/apache/airflow/compare/jason/lang-sdk-e2e/02-importer-registry...jason/lang-sdk-e2e/02b-shared-dag-cycle-detection)). - **next**: #73842 uses it to reject a native Dag whose task graph has a cycle. ## Why Cycle detection only runs for Dags that go through the Python SDK's `DAG` object. A Dag from another language SDK reaches core already serialized, so nothing rejects a cycle in it. ## How Core cannot reuse `DAG.check_cycle` as it is: it is a method on the authoring class and reads task objects, while core holds serialized data it should not hydrate just to run this check. - The traversal moves to `shared/dagnode`, which `task-sdk` and `airflow-core` already depend on. - The graph is described by callbacks, not node objects, so a caller answers from whatever data it holds. - It returns the offending task id instead of raising. `AirflowDagCycleException` lives in the Task SDK, and importing it here would invert the dependency, so each caller raises its own error. ```python from airflow._shared.dagnode.cycle import detect_cycle downstream = {"extract": ["load"], "load": ["extract"]} detect_cycle(downstream, downstream.__getitem__) # "load", or None when there is no cycle ``` ## What - `detect_cycle(node_ids, downstream_of)` in `shared/dagnode/src/airflow_shared/dagnode/cycle.py`, with its unit tests. - `DAG.check_cycle()` delegates to it and keeps its message, `Cycle detected in Dag: <dag_id>. Faulty task: <task_id>`. No behavior change: the walk still follows downstream edges only and stays iterative, and the existing `TestCycleTester` tests pass unchanged. No newsfragment, since nothing user-visible changes. ## How to test ```bash uv run --project task-sdk pytest task-sdk/tests/task_sdk/definitions/test_dag.py -k cycle -xvs ``` `shared/dagnode/tests/dagnode/test_cycle.py` covers `detect_cycle` itself. --- ##### Was generative AI tooling used to co-author this PR? - [x] Yes, with help of Claude Code (Claude Opus 4.5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) -- 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]
