This is an automated email from the ASF dual-hosted git repository.
jedcunningham pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new c1fcbefd712 Fix nested non-Dag with-statement exiting the Dag context
in version inflation checker (#68795)
c1fcbefd712 is described below
commit c1fcbefd71256cc47cafc2cb989a0ac58838035a
Author: Pradeep Kalluri <[email protected]>
AuthorDate: Fri Aug 21 16:44:26 2026 +0100
Fix nested non-Dag with-statement exiting the Dag context in version
inflation checker (#68795)
AirflowRuntimeVaryingValueChecker.visit_With() called exit_dag_context()
unconditionally at the end of every with-statement. When a non-DAG
with-statement (e.g. `with open(...) as f`) was nested inside a DAG
with-block, it would set is_in_dag_context=False, causing any tasks
constructed AFTER the nested with — but still inside the DAG block —
to be invisible to the checker. Those tasks were never flagged for
runtime-varying values even when they used datetime.now(), random(), etc.
Fix: guard the exit_dag_context() call with `if is_with_dag_context`
so only the with-statement that actually entered the DAG context exits it.
Regression tests added for:
- nested non-DAG with followed by a task (primary case)
- multiple successive nested non-DAG withs
- task constructed inside a nested non-DAG with
- task outside the DAG block not over-flagged after the fix
---
.../airflow/utils/dag_version_inflation_checker.py | 6 +-
.../utils/test_dag_version_inflation_checker.py | 77 ++++++++++++++++++++++
2 files changed, 81 insertions(+), 2 deletions(-)
diff --git a/airflow-core/src/airflow/utils/dag_version_inflation_checker.py
b/airflow-core/src/airflow/utils/dag_version_inflation_checker.py
index 2314b94a126..06fa96a7faa 100644
--- a/airflow-core/src/airflow/utils/dag_version_inflation_checker.py
+++ b/airflow-core/src/airflow/utils/dag_version_inflation_checker.py
@@ -513,8 +513,10 @@ class AirflowRuntimeVaryingValueChecker(ast.NodeVisitor):
for body in node.body:
self.visit(body)
- # Exit Dag with block
- self.dag_detector.exit_dag_context()
+ # Only exit Dag with block if we entered it; unconditional exit would
prematurely
+ # reset the context when a non-DAG with-statement is nested inside a
DAG with block.
+ if is_with_dag_context:
+ self.dag_detector.exit_dag_context()
def visit_FunctionDef(self, node: ast.FunctionDef):
for decorator in node.decorator_list:
diff --git
a/airflow-core/tests/unit/utils/test_dag_version_inflation_checker.py
b/airflow-core/tests/unit/utils/test_dag_version_inflation_checker.py
index 8d858523be4..035375c01f3 100644
--- a/airflow-core/tests/unit/utils/test_dag_version_inflation_checker.py
+++ b/airflow-core/tests/unit/utils/test_dag_version_inflation_checker.py
@@ -812,3 +812,80 @@ with DAG(
"""
warnings = self._check_code(code)
assert len(warnings) == 1
+
+ def test_nested_non_dag_with_does_not_exit_dag_context(self):
+ """Regression test: a non-DAG with-statement nested inside a DAG
with-block must not
+ prematurely exit the DAG context.
+
+ Before the fix, visit_With() called exit_dag_context() unconditionally
regardless of
+ whether a DAG context was entered. This meant any nested non-DAG
with-statement (e.g.,
+ ``with open(...) as f``) would reset is_in_dag_context=False, causing
tasks that appear
+ AFTER the nested with-statement (but still inside the DAG with-block)
to be missed.
+ """
+ code = """
+from airflow import DAG
+from datetime import datetime
+from airflow.operators.bash import BashOperator
+
+with DAG('my_dag') as dag:
+ with open('some_file.txt') as f: # non-DAG with — must not exit DAG
context
+ data = f.read()
+ # This task is still inside the DAG block and uses datetime.now() — must
be flagged
+ t1 = BashOperator(
+ task_id='test',
+ bash_command=str(datetime.now()),
+ )
+"""
+ warnings = self._check_code(code)
+ assert len(warnings) == 1, (
+ "BashOperator after a nested non-DAG with-statement must still be
detected "
+ "as inside the DAG context and flagged for using datetime.now()"
+ )
+
+ def test_multiple_nested_non_dag_withs_do_not_exit_dag_context(self):
+ """Multiple successive non-DAG with-statements inside a DAG block must
not exit the context."""
+ code = """
+from airflow import DAG
+from datetime import datetime
+from airflow.operators.bash import BashOperator
+
+with DAG('my_dag') as dag:
+ with open('a.txt') as f1:
+ pass
+ with open('b.txt') as f2:
+ pass
+ t1 = BashOperator(task_id='t', bash_command=str(datetime.now()))
+"""
+ warnings = self._check_code(code)
+ assert len(warnings) == 1
+
+ def test_task_inside_nested_non_dag_with_is_still_flagged(self):
+ """A task constructed inside a nested non-DAG with-block is still in
DAG context."""
+ code = """
+from airflow import DAG
+from datetime import datetime
+from airflow.operators.bash import BashOperator
+
+with DAG('my_dag') as dag:
+ with open('a.txt') as f:
+ t1 = BashOperator(task_id='t', bash_command=str(datetime.now()))
+"""
+ warnings = self._check_code(code)
+ assert len(warnings) == 1
+
+ def test_task_outside_dag_with_not_flagged_after_nested_fix(self):
+ """Tasks genuinely outside any DAG block must not be flagged — the fix
must not over-flag."""
+ code = """
+from airflow import DAG
+from datetime import datetime
+from airflow.operators.bash import BashOperator
+
+with DAG('my_dag') as dag:
+ with open('a.txt') as f:
+ pass
+
+# This task is outside the DAG with block — should NOT be flagged
+t_outside = BashOperator(task_id='outside', bash_command=str(datetime.now()))
+"""
+ warnings = self._check_code(code)
+ assert len(warnings) == 0, "A task constructed outside the DAG
with-block must not be flagged"
