Fix Variable.get() at parse time during dag.test()#63459
Open
andreahlert wants to merge 6 commits into
Open
Conversation
3f6d92c to
8517aa0
Compare
When running `python my_dag.py` with `dag.test()` at the bottom, any `Variable.get()` at module level fails because `SUPERVISOR_COMMS` is not yet initialized. Add direct DB query fallback in `_get_variable` and `_get_connection` (context.py) that fires only when no SUPERVISOR_COMMS exists and we are not in a server process. This avoids the Execution API HTTP layer which would cause a recursive deadlock. closes: apache#51816 Signed-off-by: André Ahlert <andre@aex.partners>
8517aa0 to
fc79672
Compare
time_machine.travel with a naive string treats it as local time, causing logical_date mismatch in UTC-offset timezones. Use UTC-aware datetime instead. Also fix positional vs keyword arg mismatch in mock assertions to match actual send() call sites. Signed-off-by: André Ahlert <andre@aex.partners>
…-dag-test # Conflicts: # task-sdk/tests/task_sdk/execution_time/test_supervisor.py Signed-off-by: André Ahlert <andre@aex.partners>
Member
|
I am not sure if this is the right way @amoghrajesh @ashb @kaxil ? |
2 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
When running
python my_dag.pywithdag.test()at the bottom, anyVariable.get()at module level fails becauseSUPERVISOR_COMMSis notyet initialized. In production the DAG processor sets it before parsing,
but the
dag.test()/ direct script execution flow does not.__getattr__intask_runner.pythat lazily createsa
ParseTimeCommswhenSUPERVISOR_COMMSis first accessed at parse timeParseTimeCommsqueries the metadata DB directly via SQLAlchemy,avoiding the Execution API HTTP layer (which would cause a recursive
deadlock: the API route detects SUPERVISOR_COMMS and delegates back to
the SDK in an infinite loop)
_AIRFLOW_PROCESS_CONTEXTsettings.Session is None)closes: #51816
Test plan
__getattr__behavior (4 tests intest_task_runner.py)ParseTimeComms.send()with mocked DB (4 tests intest_supervisor.py)test_variables.pytests pass (9 tests)Variable.get()at top level +dag.test()resolvesvariable from DB, task executes successfully with the resolved value
Was generative AI tooling used to co-author this PR?
Generated-by : Claude Code (Opus 4.6) following [the guidelines]
Used for helping on multiple tests on breeze.
(https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)