amoghrajesh commented on code in PR #70298:
URL: https://github.com/apache/airflow/pull/70298#discussion_r4044098943
##########
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:
Handled in 3f2104367f
##########
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:
Handled in 3f2104367f
--
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]
