kaxil commented on code in PR #70298:
URL: https://github.com/apache/airflow/pull/70298#discussion_r3992990468
##########
dev/registry/extract_parameters.py:
##########
@@ -391,28 +392,138 @@ def load_resumable_job_mixin() -> type | None:
return None
+# Matches an actual self.defer() call or self.deferrable attribute read, but
not
+# self.defer_for_approval(). TaskDeferred catches operators that raise it
directly instead of
Review Comment:
The regex is right, the outcome isn't. `self\.defer\(` no longer matches
`self.defer_for_approval(`, but the `_SELF_CALL_RE` walk below follows that
call into `LLMApprovalMixin.defer_for_approval`, whose fallback branch is a
literal `self.defer(HITLTrigger(...))` at `approval.py:178`. I ran the detector
at HEAD across the 1201 registered operator, sensor and transfer classes and
logged the match site for each hit: all five common.ai LLM operators come back
True on that line. `LLMOperator`, `LLMBranchOperator`, `LLMSQLQueryOperator`
and `LLMSchemaCompareOperator` reach it as `execute ->
self.defer_for_approval()`, and `LLMFileAnalysisOperator` one hop further
through `LLMOperator.execute`.
That branch is dead where the extractor runs. `defer_for_approval` takes the
`AIRFLOW_V_3_3_PLUS` path and raises `TaskAwaitingInput` with no trigger at
all, and I get `AIRFLOW_V_3_3_PLUS is True` importing it from the core the
extractor imports. The call also sits behind `if self.require_approval:`
(`llm.py:167-169`). The five still ship `supports_deferrable: true`, a
Deferrable badge whose tooltip promises another Triggerer picks the work back
up, and a slot in both filter counts.
Whichever way you land on the badge itself,
`test_defer_for_approval_does_not_false_match`
(`test_extract_parameters.py:562`) is the part worth fixing first:
`DefersForApprovalOnly.defer_for_approval` at :510 has a `return None` body, so
it passes against the stub and cannot fail against the real shape. Giving that
fixture's helper a `self.defer(` body, the way the Discord fixture puts the
real behaviour in the overriding method, makes it test the thing it's named for.
##########
registry/src/css/main.css:
##########
@@ -3297,6 +3298,26 @@ main {
max-width: 16rem;
}
+.capability-filter-toggles {
Review Comment:
This group can't shrink, so it runs past the viewport on a phone. It's
`display: flex` with no `flex-wrap`, its children are `white-space: nowrap`
(:3314), and nothing sets `min-width: 0`, so its width is pinned to its content
width however narrow the parent gets. I rendered the real bundle and measured
it: 370.1px with single-digit counts, 384.5px at `(24)`/`(18)`, 394.3px at
Amazon's scale, and still 384.5px when I force `.modules-header` down to 310px.
`.container` is 358px wide inside a 390px viewport (`--space-4` each side), so
it overflows on every provider page that renders these, worst on the big
providers.
The `flex-wrap: wrap` from the earlier round is on `.modules-header`, which
only buys the group its own line under the h2. It doesn't let the group itself
break. The same property here would. Putting the counts in the label was my
suggestion and it's what pushed the labels past the width where this stays
invisible, so apologies for the round trip.
##########
dev/registry/extract_parameters.py:
##########
@@ -391,28 +392,138 @@ def load_resumable_job_mixin() -> type | None:
return None
+# Matches an actual self.defer() call or self.deferrable attribute read, but
not
+# self.defer_for_approval(). TaskDeferred catches operators that raise it
directly instead of
+# calling self.defer() (e.g. VespaIngestOperator).
+_DEFERRAL_TOKEN_RE =
re.compile(r"self\.defer\(|self\.deferrable\b|TaskDeferred\b")
+_SELF_CALL_RE = re.compile(r"self\.([A-Za-z_][A-Za-z0-9_]*)\(")
+_SUPER_EXECUTE_RE = re.compile(r"super\(\)\.execute\(")
+# Matches the @task.* decorator idiom of naming the parent class directly
instead of using
+# super() (e.g. `AgentOperator.execute(self, context)` in common.ai's
@task.agent).
+_EXPLICIT_EXECUTE_RE = re.compile(r"\b([A-Za-z_][A-Za-z0-9_]*)\.execute\(")
+
+# To prevent infinite looping, most cases in the repo are 1-2 hops away.
+_MAX_DEFERRAL_WALK_DEPTH = 6
+
+
+def _get_method_source(cls: type, name: str) -> str | None:
+ method = getattr(cls, name, None)
+ if method is None:
+ return None
+ try:
+ return inspect.getsource(method)
+ except (OSError, TypeError):
+ return None
+
+
+def _find_owner_of_execute(mro: tuple[type, ...], start_idx: int) -> type |
None:
+ """Return the first class in `mro[start_idx:]` whose own `__dict__`
defines `execute`."""
+ for cls in mro[start_idx:]:
+ if "execute" in cls.__dict__:
+ return cls
+ return None
+
+
+def _next_execute_via_super(origin: type, current: type) -> type | None:
+ """Find what `super().execute()` resolves to from `current`, per
`origin`'s MRO.
+
+ Must use `origin`'s MRO, not `current`'s own: with multiple inheritance
(e.g. a
+ `@task.kubernetes`-built class) they diverge, and a mixin's own MRO may
have no
+ relationship to the class the chain actually needs to reach.
+ """
+ try:
+ idx = origin.__mro__.index(current)
+ except ValueError:
+ return None
+ return _find_owner_of_execute(origin.__mro__, idx + 1)
+
+
+def _next_execute_hop(origin: type, current: type, source: str) -> type | None:
+ """Find the next class in `source`'s delegation chain, via
`super().execute()` or an
+ explicit `ParentClass.execute(...)` call."""
+ if _SUPER_EXECUTE_RE.search(source):
+ return _next_execute_via_super(origin, current)
+ match = _EXPLICIT_EXECUTE_RE.search(source)
+ if match is None:
+ return None
+ name = match.group(1)
+ return next((cls for cls in origin.__mro__ if cls.__name__ == name), None)
+
+
+def _find_marker_declaring_class(cls: type) -> type | None:
+ """Return the class in `cls`'s MRO whose own body sets
`__supports_durable_execution = True`.
+
+ The lookup is per-class (`_{base.__name__}__supports_durable_execution`),
not a fixed
+ string, and only matches a class whose own `__dict__` carries the
(mangled) name;
+ inheriting the attribute value from a base doesn't count, only writing it
yourself does.
+ Leading underscores in the class name are stripped first, matching
Python's own name
+ mangling rule (`_Foo` mangles to `_Foo__x`, not `__Foo__x`).
+ """
+ for base in cls.__mro__:
+ mangled = f"_{base.__name__.lstrip('_')}__supports_durable_execution"
+ if base.__dict__.get(mangled) is True:
+ return base
+ return None
+
+
+def _delegates_execute_to(cls: type, target: type, depth: int) -> bool:
+ """Return True if `cls`'s resolved `execute()` chain reaches
`target.execute`.
+
+ Covers a class that never overrides `execute` (inherits `target.execute`
directly, e.g.
+ GKEStartPodOperator), one whose override ends in
`super().execute(context)` (e.g.
+ EksPodOperator, SparkKubernetesOperator), and one that names the parent
class directly
+ instead (e.g. `AgentOperator.execute(self, context)` in `@task.agent`).
+ """
+ owner = _find_owner_of_execute(cls.__mro__, 0)
+ remaining = depth
+ while owner is not None:
+ if owner is target:
+ return True
+ if remaining <= 0:
+ return False
+ source = _get_method_source(owner, "execute")
+ if source is None:
+ return False
+ owner = _next_execute_hop(cls, owner, source)
+ remaining -= 1
+ return False
+
+
+def _execute_chain_calls_resumable(cls: type, depth: int) -> bool:
+ """Return True if some class along `cls`'s resolved `execute()` chain
calls execute_resumable().
+
+ Same walk as `_delegates_execute_to`: a delegating override's own source
may not mention
+ `execute_resumable` even though the class it hands off to does.
+ """
+ owner = _find_owner_of_execute(cls.__mro__, 0)
+ remaining = depth
+ while owner is not None:
+ source = _get_method_source(owner, "execute")
+ if source is None:
+ return False
+ if "execute_resumable" in source:
+ return True
+ if remaining <= 0:
+ return False
+ owner = _next_execute_hop(cls, owner, source)
+ remaining -= 1
+ return False
+
+
def is_durable_capable(cls: type, resumable_mixin: type | None) -> bool:
"""Return True if a class implements durable/crash-safe execution.
Two ways to qualify:
- 1. A class-level `__supports_durable_execution = True`
- declaration (for operators that implement this directly against
- task_state_store, without ResumableJobMixin -- e.g. KubernetesPodOperator,
- AgentOperator).
- 2. Genuinely implementing ResumableJobMixin's contract.
-
- The first path deliberately looks up the class prefixed attribute
- (`_{ClassName}__supports_durable_execution`) rather than a fixed string.
- A subclass that overrides execute() itself (e.g. SparkKubernetesOperator)
- may not preserve the parent's task_state_store reconnect behavior, so the
- declaration must not be inherited -- only the exact class that wrote
- `__supports_durable_execution` in its own body qualifies this way.
-
- Inheriting the mixin alone is not sufficient for the second path: a
- complete override is inert unless execute() actually calls
- execute_resumable().
+ 1. A class-level `__supports_durable_execution = True` declaration (for
operators like
+ KubernetesPodOperator/AgentOperator that implement this directly against
+ task_state_store, without ResumableJobMixin). Inherited by a subclass that
hasn't
+ replaced the declaring class's `execute()`, whether by not overriding it
at all, or by
+ delegating back via `super().execute()`.
Review Comment:
This widens a badge that already shipped, and I can't find it called out
anywhere in the PR. Running main's `is_durable_capable` and this one over the
same 1201 classes, main marks 10 durable and this branch marks 13, with nothing
lost. The three new ones are `EksPodOperator`, `GKEStartPodOperator` and
`SparkKubernetesOperator`. The first two look like real false negatives getting
fixed, since GKE defines no `execute` of its own and Eks ends every path in
`super().execute(context)`.
`SparkKubernetesOperator` is the one I'd look at twice, because it's the
class the old docstring named as the reason the marker must not be inherited.
`_delegates_execute_to` reads any `super().execute(` in the source as
delegation, but that `execute` returns out of `execute_async(context)` before
reaching it whenever `self.deferrable` is set (`spark_kubernetes.py:357-361`).
The badge is still correct for it, just not for the reason this check tests:
what persists and reattaches the pod is `get_or_create_pod`, and both of KPO's
`execute_sync` and `execute_async` call it. So the rule is looser than the
docstring reads, and a subclass that branches away from `super()` without that
backstop would inherit the claim anyway.
No objection to the widening, it fixes two real misses. Mostly asking
whether it wants to be its own PR given it changes output for an
already-released badge, and the description's "across the full dataset ... 6
durable" figures want a pass either way.
--
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]
