Improve error handling for SUPERVISOR_COMMS access outside task context - #61630
Improve error handling for SUPERVISOR_COMMS access outside task context#61630andreahlert wants to merge 10 commits into
Conversation
There was a problem hiding this comment.
This fix looks reasonable in isolation, but several provider tests appear to have relied on the previous permissive behavior when SUPERVISOR_COMMS was absent or not initialized. By normalizing it to None, execution-time paths (particularly around connection resolution) are now enforced and fail loudly, which is causing widespread provider CI failures. It seems those tests did not anticipate SUPERVISOR_COMMS being present but None and therefore entering these stricter code paths. It appears that provider tests might have to be loosened (or maybe connection resolution migh have to check for SUPERVISOR_COMMS) to accomodate this change but I am not going to make such a sweeping suggestion unilaterally.
Thanks for the feedback. Already handled it with is_in_task_sdk_execution_context(), which checks that SUPERVISOR_COMMS is not None. Variable and Connection use it now, so provider tests keep the legacy path. |
There was a problem hiding this comment.
@andreahlert Thank you very much for working on this!
I believe this is a meaningful improvement to Airflow's error handling. The change provides a clearer error message and suggests a possible solution for end users, which is definitely helpful.
That said, my impression is that this PR does not solve the root cause of the issue discussed in #51816. Because of this, I think the PR description should probably be updated from "Fixes" to "Related", and maybe we should make the title more descriptive, this is an error handling improvement.
The core issue is that there are valid use cases where accessing Airflow Variables during DAG parsing is necessary. Examples include:
- The use case described by @opeida (the author of #51816)
- The one described by @DartVeDroid here:
#51816 (comment) - The implementation in Cosmos, which I mentioned in #51816 (comment), where it caches the dbt project graph representation as an Airflow Variable to optimise dbt project loading:
astronomer/astronomer-cosmos#1014
@kaxil confirmed that these are valid use cases here:
#51816 (comment)
Since we allow that in the actual dag parsing models (as it goes via DAG processor -> Supervisor comms -> Variable), we should do the same for dag.test.
@ashb gave a hint on a possible solution in #51816 (comment)
So now back to the main question: can we "just" make it set up an in-process API server there to try and ask the local DB in that case? Are there any security risks of doing that?
Because of this, my understanding is that the underlying problem remains unresolved, and this PR mainly improves the user-facing error message rather than fixing the behaviour itself.
|
Thanks for the detailed review, Tatiana. I'll update this PR to "Related to" instead of "Fixes" and adjust the title accordingly. For the actual fix, I'll open a separate PR implementing the lazy-init approach via |
andreahlert
changed the title
|
Friendly ping to my fellow friends: @kaxil @amoghrajesh @XD-DENG @ashb |
potiuk
added
the
ready for maintainer review
|
Moved milestone to 3.2.1 as there is no review from maintainers and I am just about to start rc2 for 3.2.0 |
|
Looks like all review comments have been addressed. @ashb and @amoghrajesh could you review again? |
|
@andreahlert — Removing the The label's contract is that the PR is ready for maintainer review — a regression like this means the PR temporarily isn't. Rebase your branch onto the latest
No rush. Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you. |
SUPERVISOR_COMMS was declared as a bare type annotation without assignment, which does not create an actual module attribute in Python. This caused ImportError when Variable.get() was called at the top level of a DAG file (outside task execution context), such as during dag.test(). Initialize SUPERVISOR_COMMS to None so the import always succeeds, add None guards before .send() calls in _set_variable, _delete_variable, and ExecutionAPISecretsBackend methods, and provide helpful error messages suggesting alternatives like environment variables or Jinja templates. Closes: apache#51816 Signed-off-by: André Ahlert <andre@aex.partners>
…ruff format Use type: ignore[assignment] instead of | None for SUPERVISOR_COMMS to avoid 39 mypy union-attr errors. Replace hasattr with getattr is not None in variable.py and connection.py. Fix reinit_supervisor_comms to check is None. Simplify set_supervisor_comms context manager. Update test assertions. Signed-off-by: André Ahlert <andre@aex.partners>
Assert only on operator logger messages so CI logs (e.g. OTLP connection errors to localhost:4318) do not break the test. Signed-off-by: André Ahlert <andre@aex.partners>
Signed-off-by: André Ahlert <andre@aex.partners>
Per review feedback: keep SUPERVISOR_COMMS as a bare type annotation (not initialized to None) and use hasattr checks in the few places that need to detect task execution context. Signed-off-by: André Ahlert <andre@aex.partners>
Signed-off-by: André Ahlert <andre@aex.partners>
70fdb01 to
5631eed
Compare
Signed-off-by: André Ahlert <andre@aex.partners>
…s-import-error Signed-off-by: André Ahlert <andre@aex.partners>
potiuk
added
the
ready for maintainer review
|
@ashb — You commented inline on this PR back in March and @tatiana called it "a meaningful improvement to Airflow's error handling". The PR has been sitting at REVIEW_REQUIRED since then with green CI and no unresolved threads. Could you convert your engagement to a formal APPROVE so it can land? Same situation as the sister PR #61627 from @andreahlert. Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Drafted-by: Claude Code (Opus 4.7); reviewed by @potiuk before posting |
|
|
||
| raise AirflowRuntimeError( | ||
| ErrorResponse(error=ErrorType.VARIABLE_NOT_FOUND, detail={"message": f"Variable {key} not found"}) | ||
| ) |
There was a problem hiding this comment.
Related: _get_variable_keys below (line 329) still has the bare from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS at the top of the function. Since SUPERVISOR_COMMS is now a bare annotation in task_runner.py (SUPERVISOR_COMMS: CommsDecoder[...] at line 930), this import raises ImportError at function entry whenever the function is called outside a task context -- exactly the opaque failure mode this PR is trying to wrap with a helpful message.
Variable.keys() from a DAG top-level (or any non-task caller) will hit the original ImportError, not the friendly message added to _get_variable here. Worth applying the same hasattr(task_runner, "SUPERVISOR_COMMS") guard there for symmetry, or switching to from airflow.sdk.execution_time import task_runner + task_runner.SUPERVISOR_COMMS like _set_variable / _delete_variable now do.
There was a problem hiding this comment.
Good catch, that one slipped through. Fixed.
| log.exception(e) | ||
|
|
||
| SUPERVISOR_COMMS.send(PutVariable(key=key, value=value, description=description)) | ||
| if not hasattr(task_runner, "SUPERVISOR_COMMS"): |
There was a problem hiding this comment.
This guard fires after the conflict-check loop above has already iterated every secrets backend and potentially emitted the log.warning("The variable %s is defined in the %s secrets backend, which takes precedence...") message -- even though we're about to bail out because there is no SUPERVISOR_COMMS to send PutVariable through.
Compare with _delete_variable just below, which guards at the top of the function before any work. For consistency (and to avoid the misleading warning about a write that never happens), hoist this hasattr check above the conflict-check loop.
| # leaking a lot of state). Only assert on the operator's logger so other loggers (e.g. OTLP trace | ||
| # export errors in CI) do not affect the test. | ||
| operator_logger_prefix = "airflow.task.operators" | ||
| operator_messages = [r.message for r in caplog.records if r.name.startswith(operator_logger_prefix)] |
There was a problem hiding this comment.
This OTLP-noise filter is unrelated to the SUPERVISOR_COMMS error-handling scope of the PR. It's a legitimate CI-flake fix (other loggers polluting caplog.messages), but bundling it here makes the change harder to bisect later if it ever causes regressions.
Either split it into its own PR or call it out explicitly in the description so reviewers know it's intentional scope creep.
There was a problem hiding this comment.
Fair, the scope creep is real. Kept it in this PR since it surfaced while debugging the CI flake that was blocking the merge, but called it out explicitly under "Out-of-scope change included" in the description, referencing commit f2f1f3c so it stays discoverable if it ever needs to be bisected.
…t() guard - _get_variable_keys: replace `from ... import SUPERVISOR_COMMS` (raises ImportError, bare annotation) with `task_runner` namespace + hasattr guard; previously Variable.keys() outside task context hit the same opaque failure this PR is trying to wrap. - _set_variable: hoist hasattr guard above the secrets-backend conflict-check loop so the misleading "API Server will be updated" warning is not emitted when the write cannot happen. Matches the pattern already applied to _get_variable and _delete_variable. Signed-off-by: André Ahlert <andre@aex.partners>
|
@andreahlert — I've removed the Automated triage note drafted by an AI-assisted tool — may get things wrong; a real Apache Airflow maintainer takes the next look once it's green. (why automated) Drafted-by: Claude Code (Opus 4.8); reviewed by @potiuk before posting |
potiuk
removed
the
ready for maintainer review
…s-import-error # Conflicts: # task-sdk/tests/task_sdk/definitions/test_variables.py Signed-off-by: André Ahlert <andre@aex.partners>
|
@andreahlert I see CI is failing can you fix this? |
|
@andreahlert there are conflicts and test failures to resolve |
* Resolve Variables and Connections in top-level Dag code Changes: - Add `_compat/parse_time.py` with `ParseTimeComms` and `parse_time_supervision`, which install a metastore-backed `SUPERVISOR_COMMS` endpoint for the duration of a Dag parse - Extract `_lookup_variable`/`_lookup_connection` hooks from `FakeSupervisorComms` so the parse-time shim reuses its response shaping - Wire `build_dag_bag(path, comms=...)` through `collection.py`, `fixtures/dagbag.py`, and `smoke.py`, the four places the plugin executes user Dag modules - Add the `airflow_parse_secrets` fixture for lookups outside a file parse - Add `--airflow-parse-secrets` and the matching ini option, `metastore` or `off` - Certify `secrets_resolution` and require `airflow.sdk.execution_time.context._get_variable`, `._get_connection`, and `cache.SecretCache` so apache/airflow#61630 fails loudly Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * Harden parse-time secrets against Airflow's swallowed lookup errors Changes: - Log the real cause of a failed lookup at `error` before re-raising it, since Airflow's worker secrets backend does a bare `except Exception: return None` and would otherwise present a downed metadata database as an ordinary unseeded row - Move the supervision block outside `build_dag_bag`'s construction `try`, so a teardown failure is not relabeled `DagBagConstructionError` over a correctly parsed DagBag - Log and swallow a failing session close in `parse_time_supervision`, and drop the session reference from `ParseTimeComms.close` in a `finally` - Filter the connection payload to the fields the release's `ConnectionResult` declares; `description` is seedable but undeclared, and the models forbid extras before 3.3.0 - Restore Airflow's secrets cache with `SecretCache.init` after resetting it, instead of leaving it disabled for the rest of the process - Initialize the metadata database before `_build_smoke_corpus` parses, so a top-level lookup does not charge the one-time migration to a smoke item's parse timeout - Log an unseeded lookup at `debug`, not `warning`; `Variable.get(key, default=...)` makes a miss an ordinary outcome - Lead the seeding guide with session-scoped seeding, since `full_dag_bag` stashes its bag for the whole worker and the function-scoped recipe only holds for the first test - Add the `0.8.0` changelog link references Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * Add run_dag fixture and lead docs with testing real Dags (#176) * Document pytest's shared-tmpdir GC race for concurrent local runs (#169) * Document pytest's shared-tmpdir GC race for concurrent local runs Changes: - Add a `## Concurrent local runs` section to `docs/development.md` describing the `FileNotFoundError: .../pytest-current` race from `_pytest/pathlib.py::cleanup_dead_symlinks`, its cause (pytest's shared `$TMPDIR/pytest-of-<user>/` numbered-directory GC, not this plugin or `AIRFLOW_HOME`), and the `TMPDIR`-per-checkout workaround along with its `AIRFLOW_HOME` storage-ladder tradeoff. - Add a `CHANGELOG.md` entry under `[Unreleased]`. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * Attribute the tmpdir race to this plugin's retention default, not just pytest Changes: - Correct `docs/development.md`'s "Concurrent local runs" section: the race is pytest's own unguarded `unlink()`, but this plugin's zero-ini `tmp_path_retention_policy = "failed"` default (pytest's own default is `all`) is what reliably dangles the `pytest-current` symlink by deleting a passing session's own numbered directory in `pytest_sessionfinish` -- without it the race essentially can't arise. - Add `pytest -o tmp_path_retention_policy=all` and `PYTEST_DEBUG_TEMPROOT` as cheaper workarounds ahead of `TMPDIR`, since neither touches the `AIRFLOW_HOME` storage ladder the way overriding `TMPDIR` does. - Update the matching `CHANGELOG.md` entry to name all three workarounds. An adversarial review round surfaced this: the doc as first drafted said "not this plugin" for causation, which undersold this plugin's own default as the actual trigger for the dangling symlink precondition. Verified directly against `_pytest/tmpdir.py::pytest_sessionfinish` and this plugin's `defaults.py`. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com> * Add a retry-behavior cookbook recipe (#170) * Add a retry-behavior cookbook recipe Changes: - Add `## Retry behavior` to `docs/guide/cookbook.md`, driving a retry-configured task through `fail -> up_for_retry -> succeed` via a second, explicit `dag_maker.run_ti(..., ignore_ti_state=True, ignore_task_deps=True)` call - Add the backing `test_run_ti_retries_a_failed_task_instance_to_success` to `tests/enduser/test_dag_run_result.py`, asserting `try_number`, `retry_delay` via `next_retry_datetime()`, and the user's `on_retry_callback` firing, all state math with no wall-clock wait - Note the CHANGELOG entry Co-Authored-By: Claude <noreply@anthropic.com> * Fix retry-recipe try_number bookkeeping for review findings Changes: - Drop the duplicated `xcom_pull` assertion in the new retry test - Bump `try_number` before both `run_ti` calls (not just the second) so it matches `Dag.test()`'s own per-attempt bookkeeping instead of trailing it by one, and assert the resulting `1`/`2` sequence - Mark the new test `requires_airflow3`: pre-2.10 Airflow exposes `try_number` as a read-only derived property rather than a plain column, so the exact-value asserts do not hold on the 2.7-2.9 compat legs - Mirror both fixes in the `docs/guide/cookbook.md` recipe Co-Authored-By: Claude <noreply@anthropic.com> --------- Co-authored-by: Claude <noreply@anthropic.com> * Add run_dag fixture and lead docs with testing real Dags Changes: - Add a `run_dag` fixture that persists an externally-authored Dag (typically pulled from `full_dag_bag`) through a full DagRun and returns the same `DagRunResult` snapshot `dag_maker.run()` does, reusing the already dag-agnostic `_compat/dag.py` persist/create/execute primitives - Add the `RunDag` protocol type alongside `RunTask`/`DagMaker` - Rework the README quickstart, `docs/index.md`, and `docs/guide/task-execution.md` so the first example loads an existing Dag via `full_dag_bag` + `run_dag`; the inline `dag_maker` example moves to a clearly labeled secondary "adhoc Dag" path - Add a "Testing a Dag defined elsewhere" walkthrough documenting the `--dag-folder`/`airflow_dags_folder` option and the same-`dag_id` cross-worker xdist constraint `run_dag` inherits from `ensure_dag_absent` Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * Fix run_dag session leak, weak test, and inaccurate xdist race docs Changes: - Close `_DagRunner`'s metadata session when `persist_dag` fails, mirroring `_DagContext.__exit__`'s existing `finally` guard; the previous unguarded call leaked the session and dropped the record from teardown tracking - Extract `_close_records` shared by `_DagFactory.close`/`_DagRunner.close`, removing the byte-for-byte duplicated cleanup loop - Strengthen `test_run_dag_passes_through_run_id_logical_date_and_dag_run_kwargs` to assert on `conf`, a field `create_dag_run` never defaults -- the previous `state=RUNNING` value was identical to the default and asserted nothing - Correct the "same dag_id across xdist workers" race description in `docs/guide/task-execution.md` and the `RunDag` docstring: the loser is not guaranteed a clean `ValueError` -- both workers can pass the absence check and silently share one bundle row, with the first to tear down deleting metadata the other's still-running test depends on Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * Colocate the smoke catalog with full_dag_bag under --dist loadgroup (#173) * Colocate the smoke catalog with full_dag_bag under --dist loadgroup Changes: - Add `_colocate_smoke_catalog_with_full_dag_bag` to `plugin.py`, forcing the synthesized smoke catalog and any `full_dag_bag` consumer onto a shared `xdist_group` whenever both exist in a run, unless an item already carries its own explicit group - Add `FULL_DAG_BAG_FIXTURE_NAME` and `FULL_DAG_BAG_XDIST_GROUP` to `fixtures/dagbag.py` - Identify the synthesized smoke items by collection-time identity diff rather than the public `smoke` marker, which user tests can also carry - Document the new co-location behavior in `docs/guide/smoke-tests.md` and `CHANGELOG.md` Closes #163. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * Fix review findings in the smoke/full_dag_bag xdist colocation Changes: - Group the catalog with only one full_dag_bag consumer, not every one, so a suite with many such consumers does not have all their execution serialized onto a single worker just to save one parse - Detect `--dist=loadgroup` via both `config.option.dist` and xdist's worker-side synthetic `config.option.loadgroup`, since `xdist.remote.setup_config` resets `dist` back to `"no"` on a real worker - Predict `-m` survival before deciding co-location, since `_pytest.mark`'s deselection hook is normal priority and runs after this plugin's `tryfirst` one -- avoids grouping the catalog with a consumer `-m` is about to drop - Gate marker addition on `--dist=loadgroup` actually being active, so `-p no:xdist` with `--strict-markers` no longer aborts on an unregistered `xdist_group` marker - Strip xdist's `--dist loadgroup` `@<group>` nodeid suffix in `record.py` before keying `--airflow-record` outcomes, so a co-located item's recorded nodeid matches the same test recorded under any other run mode - Simplify smoke-item identification to a tail slice of `items`, re-filtered against `--airflow-baseline-select`'s own deselection Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com> * Add `airflow_smoke_disable` to opt out of bundled smoke items (#171) * Add `airflow_smoke_disable` to opt out of bundled smoke items Changes: - Add `airflow_smoke_disable` ini option, a list of bundled smoke item names to drop from the catalog (`type="linelist"`, default empty) - Add `_disabled_smoke_items` reader and `_SMOKE_ITEM_NAMES` catalog constant, validating unknown names at startup - Gate every yield in `SmokeCollector.collect()` on the disabled set - Add `_smoke_serialization_needed`, deriving whether any collected item still needs a serialized Dag, and use it in `_build_smoke_corpus` and `_serialized_dag_cache` to skip calling the Airflow DAG serializer entirely once nothing needs it - Document the new option in `docs/guide/smoke-tests.md`, contrasting it with `--deselect` (which filters after the corpus is already built) Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * Fix review findings in the smoke-disable opt-out Changes: - Guard `_validate_smoke_options` on `_disabled_smoke_items(config)` unconditionally, so a malformed `airflow_smoke_disable` is always reported even on a run whose positional scoping would otherwise skip `SmokeCollector.collect()` (the only other caller) entirely - Stop `_validate_smoke_options` from rejecting `--airflow-smoke-update` combined with sampling when `test_dag_serialization_snapshot`, the only item that guard protects, is itself disabled - Make `ScheduleSanityItem` report a Dag's serialization failure itself once `test_dag_serialization_roundtrip` (its usual reporter) is disabled, instead of silently passing a Dag the scheduler can't serialize - Extract `_select_serialization_sample`, deduplicating the sample/seed/select/log block shared by `_build_smoke_corpus` and `_serialized_dag_cache` - Resolve `_get_dag_serializer()` lazily in `_build_smoke_corpus`, only once a Dag is actually selected for serialization Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com> * Cookbook: show what a dagbag + callable test misses (#172) * Cookbook: show what a dagbag + callable test misses Changes: - Add a "What a dagbag + callable test misses" section to `docs/guide/cookbook.md`, built around one realistic multi-task `ingest` Dag: task relations (trigger rules, branching, cross-task xcom), asset-triggered cross-Dag relations, depends-on-past/backfill DagRun sequences, and retry behavior (`up_for_retry`, `try_number`) - Demote the existing five recipes to a new `Community recipes` heading so the two framings don't blur together - Back all four new recipes with real, passing tests in `tests/enduser/test_cookbook_ingest.py` and `test_cookbook_digest.py` (the latter genuinely 3.x-only, added to `conftest.py`'s `collect_ignore` alongside `test_assets.py`) - Link the cookbook from README's `Why not...` section Closes #165. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * Fix review findings: 2.x-safe backfill recipe, correct db_test marker Changes: - Add `catchup=False` to the `ingest_backfill` Dag: without it, `PrevDagrunDep` takes a different branch on the Airflow 2.x family (`catchup_by_default` defaults `True` there), which never sees day one's manual run and passes the depends-on-past recipe for the wrong reason. Caught by the compat-matrix-aware adversarial review pass - Drop `pytest.mark.db_test` from the module-level `pytestmark` in `test_cookbook_ingest.py` and apply it per-function to the three `dag_maker` tests only, matching every sibling `tests/enduser/` module -- it was wrongly forcing DB init for the DB-free `run_task`-based retry test, which exists to prove no DagRun is needed - Drop the dead `ignore_ti_state=True` from the backfill rescue call; only `ignore_depends_on_past=True` does anything there - Fix cookbook prose that overstated all four recipes running through the `ingest` Dag via `dag_maker` (the retry recipe's second half uses the DB-free `run_task` fixture on a throwaway Dag) and that implied the cross-Dag recipe's producer reuses recipe 1's `ingest` Dag (it's a separate, `ingest`-shaped Dag) Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com> * Add render_task for DB-free template-field rendering (#174) * Add render_task for DB-free template-field rendering Changes: - Add `_compat.in_process.render_task_in_process`, extracting shared `RuntimeTaskInstance` construction from `run_task_in_process` into `_build_runtime_task_instance` so both share it - Add the `render_task` fixture (`fixtures/render.py`) and `RenderTask` protocol, gated on the Airflow 2.x family the same way as `run_task` - Add a `rendered(...)` matcher to `matchers.py` for one-expression assertions; it must be the left operand since `BaseOperator.__eq__` returns `False` instead of `NotImplemented` for foreign types - Deliberately skip `task.prepare_for_execution()` before rendering, unlike the real run path, so the resolved fields land on the caller's own operator object - Document `render_task` in `db-free-execution.md`, cross-link it from the cookbook's rendered-templates recipe, and list it alongside `run_task` in the 2.x-gated fixture set in `README.md`/`docs/index.md`/`AGENTS.md` Closes #118. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com> * Render onto a copy so render_task never mutates the caller's operator Changes: - Render onto a `prepare_for_execution()` copy (mirroring `task_runner.run`'s own preparation step) instead of the caller's own operator object -- a shared operator (module-level Dag, session-scoped fixture) previously went stale after the first `render_task` call, since a second render silently no-oped on a field that already held a plain string with no Jinja markup left - Extract `_installed_supervisor_comms`, a context manager shared by `run_task_in_process` and `render_task_in_process`, replacing the duplicated SUPERVISOR_COMMS save/swap/restore `try`/`finally` block - Return a `_RuntimeTaskInstanceBuild` NamedTuple from `_build_runtime_task_instance` instead of a bare `tuple[Any, Any]` - Add `ValueError: ... try_number is less than 1` to the `Raises:` sections that omitted it - Add a mapped-operator render test (`tests/compat` and `tests/enduser`, so it runs across the whole compat matrix): a mapped operator's `render_task` result is the concrete unmapped instance for `map_index`, never the mapped operator itself -- document this instead of asserting identity - Fix the `db-free-execution.md` example, which built an operator with no bound Dag and would crash; bind it and assert against the return value - Add a `test_matchers.py` case pinning the documented `rendered(...)` left-operand requirement against a fake operator with the same hostile `__eq__` shape as `BaseOperator` Found by the adversarial review round: two independent passes flagged the in-place mutation as a footgun from different angles (cross-test contamination, mapped-operator identity), and `/code-review high` converged with both on the missing `Raises:` line. The unbound-task-with-explicit-dag_id crash in `_build_runtime_task_instance` (shared with `run_task_in_process`) is a pre-existing gap, not a regression -- left alone, out of scope for #118. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com> --------- Co-authored-by: Claude <noreply@anthropic.com> --------- Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Summary
Related to #51816
SUPERVISOR_COMMSis declared as a bare type annotation (SUPERVISOR_COMMS: CommsDecoder[...]) intask_runner.py, which does not create an actual module attribute in Python. This causedImportErrorwhenVariable.get()was called at the top level of a DAG file (outside task execution context), e.g. duringdag.test().This PR improves error handling and messaging for this scenario but does not address the root cause. A separate PR will implement the actual fix using a lazy-init approach with
InProcessExecutionAPI.Changes
from airflow.sdk.execution_time import task_runnerand gate everySUPERVISOR_COMMS.send(...)behindhasattr(task_runner, "SUPERVISOR_COMMS"). Avoids the bare-annotationImportErrorwithout changing the runtime semantics for in-task callers._get_variable,_get_variable_keys,_set_variable, and_delete_variableat the top of the function so the friendly error is raised before any secrets-backend work (in particular,_set_variableno longer emits the misleading "API Server will be updated" warning when there is no SUPERVISOR_COMMS to send the write through).Variable.get/keys/set/deleteis called outside task context, suggesting alternatives (environment variables, Jinja templates, moving the call inside a task).Variable.get/set/deleteoutside task execution context.airflow.models.VariableandConnectionkeep using the samehasattr(..., "SUPERVISOR_COMMS")check assupervisor.py, so outside task context the legacy path is used and provider CI is unaffected.Files changed
context.pytask_runnernamespace + hasattr guards + enhanced error messages for get/keys/set/deleteexecution_api.pyExecutionAPISecretsBackendtest_variables.pyairflow-core/.../task_sdk_context.pyis_in_task_sdk_execution_context()shared withairflow.models.Variable/ConnectionOut-of-scope change included
airflow-core/tests/unit/serialization/test_serialized_objects.pycontains an unrelated fix fortest_logging_propogated_by_defaultflakiness: filteringcaplog.recordsby theairflow.task.operatorslogger so OTLP trace-export errors from other loggers no longer pollute the assertion. It was caught while debugging the CI flake blocking this PR. Not split into a separate PR to avoid an extra CI cycle for a one-line test scoping change; flagged here for reviewer awareness (commitf2f1f3c5c9).Test plan
test_variables.pytests pass (no regression)TestVariableOutsideTaskContexttests passVariable.get()with env var works outside task contextVariable.get()/Variable.keys()/Variable.set()/Variable.delete()raise the friendlyAirflowRuntimeError(noImportError) outside task context