kaxil commented on code in PR #70298:
URL: https://github.com/apache/airflow/pull/70298#discussion_r4025168956
##########
dev/registry/extract_parameters.py:
##########
@@ -391,28 +394,159 @@ 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 _strip_comment_lines(source: str) -> str:
+ """Drop whole-line comments so they can't be mistaken for real delegation
code.
+
+ A comment can say the opposite of what the code does (e.g. "overrides
execute rather
+ than calling super().execute()"), and a raw-text search can't tell the two
apart.
+ """
+ return "\n".join(line for line in source.splitlines() if not
line.strip().startswith("#"))
+
+
+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.
+
+ Every explicit match is tried in order, since an unrelated earlier call
(e.g.
+ `cursor.execute(...)`) can otherwise shadow the real delegation. The
matched class is
+ resolved to whoever actually owns `execute` (it may only inherit one, e.g.
+ `GKEStartPodOperator.execute(self, context)`), the same way a `super()`
hop is.
+ """
+ source = _strip_comment_lines(source)
+ if _SUPER_EXECUTE_RE.search(source):
+ return _next_execute_via_super(origin, current)
+
+ mro_by_name = {base.__name__: base for base in origin.__mro__}
+ for match in _EXPLICIT_EXECUTE_RE.finditer(source):
+ named_cls = mro_by_name.get(match.group(1))
+ if named_cls is None:
+ continue
+ owner = _find_owner_of_execute(origin.__mro__,
origin.__mro__.index(named_cls))
+ if owner is not None:
+ return owner
+ return 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
Review Comment:
The docstring note promised in the round-4 reply is not here, and the
description points readers at it twice.
Reply 4012645933 said "Added a docstring note explaining that". At this
commit `grep -niE "coincid|get_or_create_pod|backstop"` over `dev/registry/`
returns nothing, and 44bd0106 did not touch this docstring. The description
meanwhile says "for a reason the detector doesn't actually check (see the
docstring on `_delegates_execute_to`)", and its table row reads "coincidental
reasoning per the `_delegates_execute_to` docstring note". Both send a reader
here for an explanation that does not exist.
The line above is the part worth changing. `SparkKubernetesOperator` is
listed as an example of "one whose override ends in
`super().execute(context)`", which is the reading round 4 questioned. At HEAD
its `execute` (`spark_kubernetes.py:352-361`) is:
```python
if self.deferrable:
self.execute_async(context)
return
return super().execute(context)
```
On the deferrable path it returns at :359 and never reaches
`super().execute(context)` at :361. `_delegates_execute_to` matches it
regardless, because `_SUPER_EXECUTE_RE.search` at :462 is a text search with no
view of the early return. The verdict is still right, for the reason you gave
in the reply: `execute_sync` and `execute_async` reach the pod reattach
independently.
Adding the note you described would close it, naming the failure mode it
guards against: a subclass whose `execute` neither reaches the marker-declaring
class nor goes through the pod reattach, yet inherits the durable claim. The
example list above also wants `SparkKubernetesOperator` either dropped or
marked as the case that matches for a reason the check does not test, since as
written it teaches the opposite.
##########
registry/src/css/main.css:
##########
@@ -3297,6 +3298,27 @@ main {
max-width: 16rem;
}
+.capability-filter-toggles {
Review Comment:
The `flex-wrap` landed here and it does fix this group, but the overflow
moved to the sibling rather than going away.
Measured by rendering the built bundle at the widest real label counts the
extractor produces (amazon, 115 and 115), with the pre-PR bundle as the
baseline. This group's own `scrollWidth - clientWidth` is 0 at every width
tested, and at 320, 360 and 390 the toggles wrap to two rows with no page
overflow, so the round-4 ask is satisfied.
What changed is the search input beside it. `.module-search-wrapper` at
:3297-3299 sets `max-width: 16rem` and no `min-width`, so it becomes the flex
item that absorbs the squeeze once the toggles stop shrinking, while
`.capability-filter-toggle` children are `white-space: nowrap` at :3316.
```
pre-PR 320-900px : no page overflow, search input 195-256px
HEAD 44bd0106 435-470px : page overflow up to 34px, search input 54px
530-560px : page overflow up to 32px, search input 54px
```
At 54px the placeholder "Search modules..." no longer fits. These are
large-phone-landscape and small-tablet widths rather than phone widths, which
is why a check at 390px now reads clean.
Adding `min-width: 12rem` to `.module-search-wrapper` takes the narrowest
search input to 193px and leaves no width between 320 and 900 with page scroll,
measured the same way. One line if you want it.
##########
dev/registry/extract_parameters.py:
##########
@@ -530,6 +770,7 @@ def make_entry(
"provider_id": provider_id,
"provider_name": provider_name,
"supports_durable_execution": is_durable_capable(cls_or_obj,
resumable_mixin),
+ "supports_deferrable": supports_deferrable(cls_or_obj),
Review Comment:
Decorator entries never receive either capability field, because this is the
only path that goes through `make_entry`.
The decorator loop builds its own dict literal (:904-931 at this commit)
instead of calling `make_entry`, so every `@task.*` entry ships 11 keys rather
than 13 and falls back to the contract defaults. That is all 24 decorators the
extractor discovers. Two matter for this feature: `@task.agent` wraps
`AgentOperator` and `@task.kubernetes` wraps `KubernetesPodOperator`, both of
which the detector reports durable, so those cards can never show a badge or
match either filter. The docstrings added this round name `@task.agent` and
`@task.kubernetes` as the motivating cases for `_EXPLICIT_EXECUTE_RE` and
`_delegates_execute_to`, so the walk was extended for exactly this delegation
shape and the result is then dropped.
`test_all_module_fields_present` (`test_extract_parameters.py:1002-1013`)
cannot catch it. It derives the required set from `fields(Module)`, which is
right, but its fixture sets `"task-decorators": []` at :821, so the decorator
path is never exercised. Giving that fixture one decorator entry makes the test
fail until the entry either goes through `make_entry` or gains the two fields.
On the description. "This is the single durable-only case across the entire
registry" does not hold: your own verification table lists `AgentOperator` at
durable True and deferrable False two sections below, which is a second
durable-only case, and `SparkSubmitOperator` is not in the table at all. The
per-class rows themselves check out, so it is this sentence and the aggregates
that are off. For the aggregates I would regenerate rather than patch the
numbers, and it is worth checking the run for import warnings while you do: the
claimed durable figure is lower than what the pre-merge baseline produces,
which points at a run that was dropping classes on import rather than one that
is merely stale.
Noticed rather than requested, since it predates this PR and sits outside
the diff: the durable badge tooltip says "Requires durable=True (default)".
That holds for twelve of the thirteen badged classes, because
`ResumableJobMixin` defaults `durable=True` (`resumablejobmixin.py:99`) and
`KubernetesPodOperator` resolves `None` to `True` on 3.3+ (`pod.py:417-420`).
`AgentOperator` declares `durable: bool = False` (`agent.py:267`), so for that
one the tooltip tells a reader the opposite of what they need to do.
##########
dev/registry/extract_parameters.py:
##########
@@ -421,15 +555,121 @@ def is_durable_capable(cls: type, resumable_mixin: type
| None) -> bool:
if inspect.isabstract(cls):
return False
- execute = getattr(cls, "execute", None)
- if execute is None:
- return False
+ return _execute_chain_calls_resumable(cls, _MAX_DEFERRAL_WALK_DEPTH)
+
+
+def _is_terminal_block(body: list[ast.stmt]) -> bool:
+ """Return True if `body`'s last statement always exits the function (raise
or return)."""
+ return bool(body) and isinstance(body[-1], (ast.Raise, ast.Return))
+
+
+def _strip_dead_version_guard_branches(source: str, resolve_global:
typing.Callable[[str], object]) -> str:
+ """Blank out code after a terminal `if <flag>: raise ...` whose flag
resolves True here.
+
+ Some operators write `if AIRFLOW_V_3_3_PLUS: raise ...` with no `else`,
followed by a
+ pre-3.3 `self.defer(...)` fallback that never runs on the core being
imported.
+ `resolve_global` looks up the flag by name rather than a fixed list.
+ """
+ try:
+ tree = ast.parse(textwrap.dedent(source))
+ except SyntaxError:
+ return source
+
+ if not tree.body or not isinstance(tree.body[0], (ast.FunctionDef,
ast.AsyncFunctionDef)):
+ return source
+
+ dead_from: int | None = None
+ for stmt in tree.body[0].body:
+ if (
+ isinstance(stmt, ast.If)
+ and not stmt.orelse
+ and isinstance(stmt.test, ast.Name)
+ and _is_terminal_block(stmt.body)
+ and resolve_global(stmt.test.id) is True
+ ):
+ dead_from = stmt.end_lineno
+ break
+
+ if dead_from is None:
+ return source
+ return "".join(source.splitlines(keepends=True)[:dead_from])
+
+
+def _get_reachable_method_source(cls: type, name: str) -> str | None:
+ """Like `_get_method_source`, but with dead version-guard branches
stripped first."""
+ method = getattr(cls, name, None)
+ if method is None:
+ return None
try:
- source = inspect.getsource(execute)
+ source = inspect.getsource(method)
except (OSError, TypeError):
+ return None
+
+ # unwrap() undoes a functools.wraps() decorator, whose __globals__ would
otherwise
+ # point at the decorator's own module instead of the method's.
+ func = inspect.unwrap(getattr(method, "__func__", method))
Review Comment:
Half of the round-4 fix depends on this `unwrap`, and nothing tests it.
Neutering `inspect.unwrap` here leaves the whole suite green at 88 passed,
yet it is the only reason four of the nine classes come back False.
`BaseOperatorMeta._apply_defaults` wraps `HITLOperator.execute`, and the
wrapper's `__globals__` does not carry `AIRFLOW_V_3_3_PLUS` while the unwrapped
function's does. Both report the same `__module__`, which is why reading
`__module__` would not have surfaced it. Without this line the guard flag
resolves to nothing, the dead branch is not stripped, and `HITLOperator`,
`ApprovalOperator`, `HITLBranchOperator` and `HITLEntryOperator` all go back to
True. The five common.ai classes are unaffected, since `defer_for_approval` is
not wrapped.
The reason no test covers it is structural rather than an oversight: the
file contains no `functools`, no `wraps`, no `__globals__` and no `unwrap`
anywhere, so every fixture is a plain method and none can reach this line. The
two fixtures added this round do genuinely constrain the stripper, and
reverting that half turns exactly
`test_defer_for_approval_helper_with_dead_fallback_does_not_false_match` and
`test_hitl_shaped_dead_fallback_does_not_false_match` red, so that half is
protected. This half is not.
A fixture with a `functools.wraps` wrapper whose `__globals__` belong to a
different module than the wrapped function would pin it, and it is worth
having, because a silent regression here puts back the exact nine-class false
positive this round was about.
Worth knowing that this is not specific to the 3.3 flag. `BaseOperatorMeta`
wraps `execute` on nearly every registered class, so the `__globals__` you get
back is usually not the method's own, and dozens of classes carry a version
flag that only becomes visible after unwrapping, across several
`AIRFLOW_V_3_*_PLUS` names rather than just this one. `_is_terminal_block` has
the same shape of gap: forcing it to return True also leaves the suite green,
though unlike this line it changes no verdict in the tree today.
##########
dev/registry/extract_parameters.py:
##########
@@ -421,15 +555,121 @@ def is_durable_capable(cls: type, resumable_mixin: type
| None) -> bool:
if inspect.isabstract(cls):
return False
- execute = getattr(cls, "execute", None)
- if execute is None:
- return False
+ return _execute_chain_calls_resumable(cls, _MAX_DEFERRAL_WALK_DEPTH)
+
+
+def _is_terminal_block(body: list[ast.stmt]) -> bool:
+ """Return True if `body`'s last statement always exits the function (raise
or return)."""
+ return bool(body) and isinstance(body[-1], (ast.Raise, ast.Return))
+
+
+def _strip_dead_version_guard_branches(source: str, resolve_global:
typing.Callable[[str], object]) -> str:
+ """Blank out code after a terminal `if <flag>: raise ...` whose flag
resolves True here.
+
+ Some operators write `if AIRFLOW_V_3_3_PLUS: raise ...` with no `else`,
followed by a
+ pre-3.3 `self.defer(...)` fallback that never runs on the core being
imported.
+ `resolve_global` looks up the flag by name rather than a fixed list.
+ """
+ try:
+ tree = ast.parse(textwrap.dedent(source))
+ except SyntaxError:
+ return source
+
+ if not tree.body or not isinstance(tree.body[0], (ast.FunctionDef,
ast.AsyncFunctionDef)):
+ return source
+
+ dead_from: int | None = None
+ for stmt in tree.body[0].body:
+ if (
+ isinstance(stmt, ast.If)
+ and not stmt.orelse
+ and isinstance(stmt.test, ast.Name)
+ and _is_terminal_block(stmt.body)
+ and resolve_global(stmt.test.id) is True
+ ):
+ dead_from = stmt.end_lineno
+ break
+
+ if dead_from is None:
+ return source
+ return "".join(source.splitlines(keepends=True)[:dead_from])
+
+
+def _get_reachable_method_source(cls: type, name: str) -> str | None:
+ """Like `_get_method_source`, but with dead version-guard branches
stripped first."""
+ method = getattr(cls, name, None)
+ if method is None:
+ return None
try:
- source = inspect.getsource(execute)
+ source = inspect.getsource(method)
except (OSError, TypeError):
+ return None
+
+ # unwrap() undoes a functools.wraps() decorator, whose __globals__ would
otherwise
+ # point at the decorator's own module instead of the method's.
+ func = inspect.unwrap(getattr(method, "__func__", method))
+ module_globals = getattr(func, "__globals__", None)
+ if module_globals is None:
+ return source
+ return _strip_dead_version_guard_branches(source, module_globals.get)
+
+
+def _references_deferral(
+ origin: type, current: type, source: str, visited: set[tuple[int, str]],
depth: int
+) -> bool:
+ """Return True if `source` (the resolved `execute()` of `current`, called
on `origin`) references deferral.
+
+ `origin` stays fixed across recursion so `super().execute()` hops resolve
against its
+ real MRO, while `current` walks forward through the chain.
+ """
+ if _DEFERRAL_TOKEN_RE.search(source):
Review Comment:
Grouped into one comment, since none of these changes a badge today.
This search runs on source with comments and docstrings intact:
`_get_reachable_method_source` strips dead version branches but not comments,
while the delegation path strips them at :461 via `_strip_comment_lines`, whose
docstring names this exact hazard. So a comment mentioning `self.defer()`, a
docstring `:raises TaskDeferred:`, an `except TaskDeferred`, or a plain
`self.deferrable = False` write all match. Instrumenting the regex to record
whether the deciding hit sat inside a comment or string gives zero across the
registered set, so nothing is misbadged. The shapes are in the tree, in
`MSGraphAsyncOperator.execute_complete` (`msgraph.py:238`) and at
`pod.py:1165`, but both sit in resume-side methods on classes that defer for
real.
At :646 the helper is resolved against `current`, while `self.<name>()`
dispatches on the instance's real class, which is `origin`. `current` advances
as `super().execute()` hops are followed, so past the first hop the walk reads
an ancestor's copy and cannot see a more derived override. The seeding also
differs from the siblings: this walk starts at `_references_deferral(cls, cls,
...)` (:669) while `_delegates_execute_to` (:500) and
`_execute_chain_calls_resumable` (:521) start at
`_find_owner_of_execute(cls.__mro__, 0)`. Resolving helpers against `origin`
across the dataset flips zero verdicts, and the obvious carriers miss it:
`EksPodOperator` and `GKEStartPodOperator` both override `invoke_defer_method`,
but `KubernetesPodOperator.execute` contains `if not self.deferrable:`
(`pod.py:803`), so the match at :625 returns True on that frame before the
helper loop runs.
`visited` is marked at :634 and :645 without recording the depth it was
searched at, so a node first reached shallow is skipped when re-reached with
more budget. That precondition fires 22 times on real data and still flips zero
verdicts, because the walk never bottoms out at `depth <= 0`:
`_MAX_DEFERRAL_WALK_DEPTH = 6` does not bind on any real chain. That is also
why reproducing it requires forcing a chain deeper than 6, and why it becomes
live if that constant is raised or a provider grows a deeper chain.
Fixes are one-liners each: call `_strip_comment_lines` inside
`_get_reachable_method_source`, tighten `TaskDeferred\b` to `raise
TaskDeferred` (which keeps the `VespaIngestOperator` case your comment at
:396-398 cites, `vespa_ingest.py:85`), resolve helpers against `origin`, and
key `visited` on depth. Worth one test either way: `TestIsDurableCapable` has
`test_comment_mentioning_delegation_is_not_mistaken_for_delegation` at
`test_extract_parameters.py:369`, and `TestSupportsDeferrable` (:584-651) has
fourteen tests with no comment or docstring case.
##########
dev/registry/tests/test_extract_parameters.py:
##########
@@ -266,12 +322,335 @@ def
test_manual_durable_marker_qualifies_without_mixin(self):
def test_subclass_not_redeclaring_marker_disqualifies(self):
assert is_durable_capable(ManuallyDurableSubclass,
FakeResumableJobMixin) is False
+ def test_subclass_with_no_execute_override_inherits_marker(self):
+ assert is_durable_capable(ManuallyDurableSubclassNoOverride,
FakeResumableJobMixin) is True
+
+ def test_subclass_delegating_via_super_execute_qualifies(self):
+ assert is_durable_capable(ManuallyDurableSubclassDelegating,
FakeResumableJobMixin) is True
+
+ def test_subclass_delegating_via_super_execute_multi_hop_qualifies(self):
+ assert is_durable_capable(ManuallyDurableSubclassDelegatingMultiHop,
FakeResumableJobMixin) is True
+
+ def test_multiple_inheritance_decorator_mixin_qualifies(self):
+ assert is_durable_capable(DecoratedDurableSubclass,
FakeResumableJobMixin) is True
+
+ def test_multiple_inheritance_no_own_execute_qualifies(self):
+ assert is_durable_capable(DecoratedDurableSubclassNoOwnExecute,
FakeResumableJobMixin) is True
+
+ def test_mixin_subclass_delegating_via_super_execute_qualifies(self):
+ class MixinSubclassDelegating(FullyImplementedResumableOperator):
+ def execute(self, context):
+ return super().execute(context)
+
+ assert is_durable_capable(MixinSubclassDelegating,
FakeResumableJobMixin) is True
+
+ def test_explicit_parent_class_delegation_qualifies(self):
+ assert is_durable_capable(ExplicitParentDelegatingSubclass,
FakeResumableJobMixin) is True
+
+ def
test_mixin_subclass_delegating_via_explicit_parent_call_qualifies(self):
+ class
MixinSubclassExplicitDelegating(FullyImplementedResumableOperator):
+ def execute(self, context):
+ return FullyImplementedResumableOperator.execute(self, context)
+
+ assert is_durable_capable(MixinSubclassExplicitDelegating,
FakeResumableJobMixin) is True
+
+ def test_declaring_class_with_leading_underscore_is_found(self):
+ """Python strips leading underscores from the class name when
mangling, so the
+ lookup must too, or a declaring class like `_FooOperator` is never
found."""
+
+ class _UnderscoreOperator:
+ __supports_durable_execution = True
+
+ def execute(self, context):
+ return None
+
+ assert is_durable_capable(_UnderscoreOperator, FakeResumableJobMixin)
is True
+
+ def
test_comment_mentioning_delegation_is_not_mistaken_for_delegation(self):
+ """A comment can say the opposite of what the code does; a raw-text
search must not
+ be fooled by it into reporting capable."""
+
+ class CommentedNonDelegatingSubclass(ManuallyDurableOperator):
+ def execute(self, context):
+ # overrides execute rather than calling super().execute(),
+ return None
+
+ assert is_durable_capable(CommentedNonDelegatingSubclass,
FakeResumableJobMixin) is False
+
+ def test_decoy_execute_call_does_not_shadow_real_delegation(self):
Review Comment:
This passes for a different reason than its name, so it does not currently
bind the behaviour it describes.
The decoy is `cursor.execute("select 1")`, and `cursor` is a local variable
rather than a class in the MRO. `_next_execute_hop` builds `mro_by_name` from
`origin.__mro__` class names, so `mro_by_name.get("cursor")` returns None and
the loop takes the `continue` at :468. It never reaches the
try-every-match-in-order behaviour that the docstring at :456-458 describes,
which is the thing the test is named for. Changing that loop to return on the
first resolved match leaves the whole suite green.
A decoy that names a class which is in the MRO but owns no `execute`, placed
before the real delegation, would make it bind.
--
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]
