kaxil commented on code in PR #70298:
URL: https://github.com/apache/airflow/pull/70298#discussion_r3980597453
##########
dev/registry/extract_parameters.py:
##########
@@ -432,6 +433,41 @@ def is_durable_capable(cls: type, resumable_mixin: type |
None) -> bool:
return "execute_resumable" in source
+# Hand verified set that contains classes where self.defer()/self.deferrable
is reachable only
+# through a helper method one call away from execute() (e.g.
TriggerDagRunOperator
+# delegates to _trigger_dag_af_2()), so the source grep below can't find it.
+# Add an entry only after confirming self.defer() is genuinely reachable.
+_DEFERRABLE_EXCEPTIONS = {
+
"airflow.providers.standard.operators.trigger_dagrun.TriggerDagRunOperator",
+}
+
+
+def supports_deferrable(cls: type) -> bool:
+ """Return True if the class's resolved execute() actually references
deferral.
+
+ Checking for a `deferrable` constructor parameter isn't enough: a subclass
can
+ inherit the parameter while overriding execute() with code that never
reads it,
+ and an operator that always defers unconditionally has no parameter
+ to find at all. Resolving execute() via getattr and checking its own
source for
+ self.deferrable or self.defer answers "does this code path use deferral"
+ directly, instead of proxying through whether a setting merely exists
somewhere
+ in the class hierarchy.
+ """
+ qualified_name = f"{cls.__module__}.{cls.__qualname__}"
+ if qualified_name in _DEFERRABLE_EXCEPTIONS:
+ return True
+
+ execute = getattr(cls, "execute", None)
+ if execute is None:
+ return False
+ try:
+ source = inspect.getsource(execute)
+ except (OSError, TypeError):
+ return False
+
+ return "self.deferrable" in source or "self.defer" in source
Review Comment:
Ran this against the tree (1292 registered operator, sensor and transfer
classes): `DiscordWebhookOperator` stays False, so the false positive is gone,
but the exception list covers one of four delegation shapes and eleven
registered classes that defer come back False:
- `raise TaskDeferred(` with no `self.defer` anywhere:
`VespaIngestOperator.execute` (`vespa_ingest.py:85`, `vespa/provider.yaml:53`).
`self.defer` is already a prefix of `self.deferrable`, so the first clause on
this line is dead; `"TaskDeferred" in source` is the one worth having.
- A child `execute` wrapping `super().execute(context)`: `EksPodOperator`
(`eks.py:1244`; the branch is in KPO's `execute` at `pod.py:803`) and the
deprecated `GlueCrawlerOperator` (`glue_crawler.py:317` into
`GlueCrawlerRunOperator.execute` at `:182`). Following `super().execute(` up
the MRO fixes both without touching Discord, whose `execute` never calls super.
- `self.defer` behind a helper: `DatabricksStartWarehouseOperator` and
`DatabricksStopWarehouseOperator` (`warehouse.py:183`, `:209` into
`_wait_or_defer` at `:117-118`); `DatabricksTaskBaseOperator` via
`monitor_databricks_job` to `_defer_on_run` (`databricks.py:2197`,
`:2185-2186`, `:2071`), inherited unchanged by `DatabricksNotebookOperator` and
`DatabricksTaskOperator`;
`TeradataComputeClusterProvision/Resume/SuspendOperator` via
`_compute_cluster_execute` to `_handle_cc_status`
(`teradata_compute_cluster.py:168`, unconditional).
With the count in the label the misses are visible: the Databricks page
renders `(4)` at HEAD where your round-2 screenshot showed `(7)`, and the PR
description's own "deferrable only" list (`DatabricksNotebookOperator`,
`DatabricksTaskBaseOperator`, `DatabricksTaskOperator`) is False at HEAD. The
description also still describes the constructor-parameter detector and
`get_params_from_class`, so it wants a refresh once this settles.
A bounded walk from the resolved `execute` (follow `self.<name>(` through
`getattr`, follow `super().execute(` to the next `execute` in the MRO, stop on
`self.defer`/`self.deferrable`/`TaskDeferred`) goes from 243 to 254 True across
those 1292 classes, the eleven above exactly, with `TriggerDagRunOperator`
found without the list and Discord still False since its only self-call is
`self.hook.execute()`. If the list stays instead, it needs the ten call-chain
names and a test that fails when a listed class starts matching the grep.
Separately, match `self.defer` on a word boundary: `self.defer_for_approval(`
in the five common.ai LLM operators (`llm.py:169`) counts today by prefix
accident. Whether they should carry the badge is its own call, since on 3.3+
that path raises `TaskAwaitingInput` with no trigger at all
(`approval.py:168-172`) and only pre-3.3 cores fall back to
`self.defer(HITLTrigger)`, so the Triggerer promise in the tooltip is wrong for
them on current cores.
##########
dev/registry/extract_parameters.py:
##########
@@ -463,7 +499,7 @@ def discover_classes_from_provider(
"""Discover classes from a single provider by importing its modules at
runtime.
Reads the provider.yaml to find which modules/classes to inspect, imports
them,
- and returns metadata for each discovered class with all 12 Module fields.
+ and returns metadata for each discovered class with all 13 Module fields.
Review Comment:
Ash's nit about the field count landed on `make_entry` but not here or on
`test_has_all_13_fields` (`test_extract_parameters.py:362`); both bump every
time a field is added.
##########
registry/src/provider-version.njk:
##########
@@ -403,6 +427,9 @@ eleventyComputed:
{% if module.supports_durable_execution %}
<span class="durable-badge" title="Reconnects to an
already-running job on retry instead of resubmitting. Requires durable=True
(default) and depends on operator configuration.">Durable</span>
{% endif %}
+ {% if module.supports_deferrable %}
+ <span class="deferrable-badge" title="Frees the worker slot
while waiting. If the Triggerer crashes, another one picks it up using the same
trigger class and arguments. Requires deferrable=True.">Deferrable</span>
Review Comment:
With the new predicate the badge now lands on operators that defer
unconditionally and have no `deferrable` parameter, so "Requires
deferrable=True" is an instruction that fails for them. `DateTimeSensorAsync`
(`date_time.py:106-114`) and `HITLOperator` (`hitl.py:199-244`) forward
`**kwargs` to BaseOperator, which raises `TypeError("Invalid arguments were
passed ...")` on the leftover key (`operator.py:1103-1107`); `TimeSensorAsync`
hard-codes `super().__init__(deferrable=True, **kwargs)` (`time.py:128`), so
passing it again is a duplicate-keyword TypeError. Something like "Pass
deferrable=True where the operator exposes it; some defer unconditionally"
covers both, and the same phrase sits on the filter tooltip at :402.
##########
registry/src/css/tokens.css:
##########
@@ -96,6 +97,8 @@
--color-rose-600: #e11d48;
--color-teal-400: #2dd4bf;
--color-teal-500: #14b8a6;
+ --color-teal-600: #0d9488;
Review Comment:
`--color-teal-600` was added for the previous round and nothing references
it now that the badge moved to `teal-700`; it can go.
--
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]
