amoghrajesh commented on code in PR #70298:
URL: https://github.com/apache/airflow/pull/70298#discussion_r4044081380
##########
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:
Added the note. It now says the `super().execute(` check is a text search
with no view of control flow, names `SparkKubernetesOperator` as the live case
that returns before reaching that call, explains the verdict is still right
because `execute_sync` and `execute_async` both reach the pod reattach
independently, and states the gap: a subclass that reaches neither the marker
class nor that reattach would inherit the claim without earning it. Also
dropped `SparkKubernetesOperator` from the `super().execute()` example list,
since as written it taught the opposite.
##########
registry/src/css/main.css:
##########
@@ -3297,6 +3298,27 @@ main {
max-width: 16rem;
}
+.capability-filter-toggles {
Review Comment:
Added min-width: 12rem to `.module-search-wrapper`, so it stops being the
item that absorbs the squeeze once the toggles stop shrinking.
--
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]
