amoghrajesh commented on code in PR #70298:
URL: https://github.com/apache/airflow/pull/70298#discussion_r4044090184
##########
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:
Took all four. `_get_reachable_method_source` now strips comments, so that
path matches the delegation path. `TaskDeferred\b` is now `raise TaskDeferred`,
which keeps `VespaIngestOperator` and drops `except TaskDeferred` and `:raises
TaskDeferred:`. Helpers resolve against origin instead of current, matching how
self.<name>() actually dispatches. And visited keys now include the remaining
depth, so a node first reached shallow is not skipped when re-reached with more
budget.
Added a comment-only deferral fixture to TestSupportsDeferrable, and
confirmed it binds: removing the comment stripping turns that test red.
--
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]
