This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new df265782d96 [v3-3-test] Surface durable execution badge in Airflow
Registry (#69651) (#69672)
df265782d96 is described below
commit df265782d96eb8721789ae970ff88e5c31dce4fa
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Jul 13 02:28:43 2026 +0200
[v3-3-test] Surface durable execution badge in Airflow Registry (#69651)
(#69672)
* Stage 1: Durable execution detection
* Stage 2: Adding supports_durable_execution to schema
* Stage 3: Wiring up with eleventy to show it on the UI
(cherry picked from commit 59a54336481636dde9eeb3059b1cc76c7f73fccf)
Co-authored-by: Amogh Desai <[email protected]>
---
dev/registry/extract_parameters.py | 44 +++-
dev/registry/registry_contract_models.py | 1 +
dev/registry/tests/test_extract_parameters.py | 223 ++++++++++++++++++++-
.../tests/test_registry_contract_models.py | 25 +++
registry/pnpm-workspace.yaml | 4 +-
registry/src/css/main.css | 13 ++
registry/src/provider-version.njk | 7 +-
7 files changed, 312 insertions(+), 5 deletions(-)
diff --git a/dev/registry/extract_parameters.py
b/dev/registry/extract_parameters.py
index 565057541df..0e9eff89a4d 100644
--- a/dev/registry/extract_parameters.py
+++ b/dev/registry/extract_parameters.py
@@ -88,6 +88,7 @@ class Module:
category: str
provider_id: str
provider_name: str
+ supports_durable_execution: bool
def get_category(integration_name: str) -> str:
@@ -373,16 +374,51 @@ def _get_source_line(cls: type) -> int | None:
return None
+def load_resumable_job_mixin() -> type | None:
+ """Import ResumableJobMixin for durable-execution capability checks, or
None if unavailable."""
+ try:
+ from airflow.sdk import ResumableJobMixin
+
+ return ResumableJobMixin
+ except ImportError:
+ log.warning("Could not import ResumableJobMixin")
+ return None
+
+
+def is_durable_capable(cls: type, resumable_mixin: type | None) -> bool:
+ """Return True if a class fully implements ResumableJobMixin's
crash-recovery contract.
+
+ Inheriting the mixin is not sufficient: a complete override is inert unless
+ execute() actually calls execute_resumable().
+ """
+ if resumable_mixin is None or resumable_mixin not in cls.__mro__:
+ return False
+
+ if inspect.isabstract(cls):
+ return False
+
+ execute = getattr(cls, "execute", None)
+ if execute is None:
+ return False
+ try:
+ source = inspect.getsource(execute)
+ except (OSError, TypeError):
+ return False
+
+ return "execute_resumable" in source
+
+
def discover_classes_from_provider(
provider_yaml_path: Path,
base_classes: dict[str, type],
+ resumable_mixin: type | None = None,
inventory: dict[str, str] | None = None,
version: str = "",
) -> list[dict]:
"""Discover classes from a single provider by importing its modules at
runtime.
Reads the provider.yaml to find which modules/classes to inspect, imports
them,
- and returns metadata for each discovered class with all 11 Module fields.
+ and returns metadata for each discovered class with all 12 Module fields.
"""
with open(provider_yaml_path) as f:
provider_yaml = yaml.safe_load(f)
@@ -431,7 +467,7 @@ def discover_classes_from_provider(
category: str = "",
transfer_desc: str | None = None,
) -> dict:
- """Build a full module entry dict with all 11 fields."""
+ """Build a full module entry dict with all 12 fields."""
module_name = module_path.split(".")[-1]
docstring = _get_first_docstring_line(cls_or_obj)
short_desc = docstring or transfer_desc or f"{integration}
{module_type}".strip()
@@ -448,6 +484,7 @@ def discover_classes_from_provider(
"category": category or get_category(integration),
"provider_id": provider_id,
"provider_name": provider_name,
+ "supports_durable_execution": is_durable_capable(cls_or_obj,
resumable_mixin),
}
discovered: list[dict] = []
@@ -889,6 +926,8 @@ def _main_discover(
base_classes = load_base_classes()
print(f"Loaded {len(base_classes)} base classes: {',
'.join(sorted(base_classes))}")
+ resumable_mixin = load_resumable_job_mixin()
+
# Load all provider.yaml data and map provider_id -> yaml dict / path
provider_yamls_by_id: dict[str, dict] = {}
provider_paths_by_id: dict[str, Path] = {}
@@ -923,6 +962,7 @@ def _main_discover(
discovered = discover_classes_from_provider(
yaml_path,
base_classes,
+ resumable_mixin,
inventory=inventories.get(pid),
version=version,
)
diff --git a/dev/registry/registry_contract_models.py
b/dev/registry/registry_contract_models.py
index 119a0227c1f..7295cfb4036 100644
--- a/dev/registry/registry_contract_models.py
+++ b/dev/registry/registry_contract_models.py
@@ -110,6 +110,7 @@ class ModuleContract(BaseModel):
category: str
provider_id: str | None = None
provider_name: str | None = None
+ supports_durable_execution: bool = False
class ModulesCatalogContract(BaseModel):
diff --git a/dev/registry/tests/test_extract_parameters.py
b/dev/registry/tests/test_extract_parameters.py
index bfa3b05b56f..be0f608e17e 100644
--- a/dev/registry/tests/test_extract_parameters.py
+++ b/dev/registry/tests/test_extract_parameters.py
@@ -18,11 +18,14 @@
from __future__ import annotations
+import abc
+import builtins
import json
import types
from unittest.mock import patch
import pytest
+import yaml
from extract_parameters import (
Module,
_get_source_line,
@@ -31,6 +34,8 @@ from extract_parameters import (
compare_with_ast,
discover_classes_from_provider,
get_category,
+ is_durable_capable,
+ load_resumable_job_mixin,
)
@@ -101,11 +106,147 @@ class TestGetSourceLine:
assert _get_source_line(DynamicClass) is None
+# ---------------------------------------------------------------------------
+# load_resumable_job_mixin
+# ---------------------------------------------------------------------------
+class TestLoadResumableJobMixin:
+ def test_returns_none_when_airflow_sdk_unavailable(self):
+ real_import = builtins.__import__
+
+ def fake_import(name, *args, **kwargs):
+ if name == "airflow.sdk":
+ raise ImportError("no airflow.sdk here")
+ return real_import(name, *args, **kwargs)
+
+ with patch("builtins.__import__", side_effect=fake_import):
+ assert load_resumable_job_mixin() is None
+
+
+# ---------------------------------------------------------------------------
+# is_durable_capable
+# ---------------------------------------------------------------------------
+class FakeResumableJobMixin(abc.ABC):
+ """Stand-in for airflow.sdk.ResumableJobMixin's abstract-method
contract."""
+
+ @abc.abstractmethod
+ def submit_job(self, context):
+ raise NotImplementedError
+
+ @abc.abstractmethod
+ def get_job_status(self, external_id, context):
+ raise NotImplementedError
+
+ @abc.abstractmethod
+ def is_job_active(self, status):
+ raise NotImplementedError
+
+ @abc.abstractmethod
+ def is_job_succeeded(self, status):
+ raise NotImplementedError
+
+ @abc.abstractmethod
+ def poll_until_complete(self, external_id, context):
+ raise NotImplementedError
+
+ @abc.abstractmethod
+ def get_job_result(self, external_id, context):
+ raise NotImplementedError
+
+ def execute_resumable(self, context):
+ raise NotImplementedError
+
+
+class FullyImplementedResumableOperator(FakeResumableJobMixin):
+ def execute(self, context):
+ return self.execute_resumable(context)
+
+ def submit_job(self, context):
+ return "job-1"
+
+ def get_job_status(self, external_id, context):
+ return "RUNNING"
+
+ def is_job_active(self, status):
+ return status == "RUNNING"
+
+ def is_job_succeeded(self, status):
+ return status == "SUCCEEDED"
+
+ def poll_until_complete(self, external_id, context):
+ return None
+
+ def get_job_result(self, external_id, context):
+ return None
+
+
+class PartiallyImplementedResumableOperator(FakeResumableJobMixin):
+ """Retry-path methods left unoverridden -- would only blow up on an actual
crash-recovery retry."""
+
+ def execute(self, context):
+ return self.execute_resumable(context)
+
+ def submit_job(self, context):
+ return "job-1"
+
+ def poll_until_complete(self, external_id, context):
+ return None
+
+ def get_job_result(self, external_id, context):
+ return None
+
+
+class UnwiredResumableOperator(FakeResumableJobMixin):
+ """execute() never calls execute_resumable -- dead capability, never
exercised."""
+
+ def execute(self, context):
+ return self.submit_job(context)
+
+ def submit_job(self, context):
+ return "job-1"
+
+ def get_job_status(self, external_id, context):
+ return "RUNNING"
+
+ def is_job_active(self, status):
+ return status == "RUNNING"
+
+ def is_job_succeeded(self, status):
+ return status == "SUCCEEDED"
+
+ def poll_until_complete(self, external_id, context):
+ return None
+
+ def get_job_result(self, external_id, context):
+ return None
+
+
+class PlainOperator:
+ def execute(self, context):
+ return None
+
+
+class TestIsDurableCapable:
+ def test_fully_implemented_and_wired_qualifies(self):
+ assert is_durable_capable(FullyImplementedResumableOperator,
FakeResumableJobMixin) is True
+
+ def test_missing_retry_path_overrides_disqualifies(self):
+ assert is_durable_capable(PartiallyImplementedResumableOperator,
FakeResumableJobMixin) is False
+
+ def test_implemented_but_not_called_from_execute_disqualifies(self):
+ assert is_durable_capable(UnwiredResumableOperator,
FakeResumableJobMixin) is False
+
+ def test_no_mixin_in_mro_disqualifies(self):
+ assert is_durable_capable(PlainOperator, FakeResumableJobMixin) is
False
+
+ def test_mixin_unavailable_disqualifies(self):
+ assert is_durable_capable(FullyImplementedResumableOperator, None) is
False
+
+
# ---------------------------------------------------------------------------
# Module dataclass
# ---------------------------------------------------------------------------
class TestModuleDataclass:
- def test_has_all_11_fields(self):
+ def test_has_all_12_fields(self):
m = Module(
id="amazon-s3-S3Hook",
name="S3Hook",
@@ -118,6 +259,7 @@ class TestModuleDataclass:
category="amazon-s3",
provider_id="amazon",
provider_name="Amazon",
+ supports_durable_execution=False,
)
assert m.id == "amazon-s3-S3Hook"
assert m.provider_name == "Amazon"
@@ -534,6 +676,85 @@ class TestSensorNotClassifiedAsOperator:
assert result[0]["name"] == "MySensor"
+# ---------------------------------------------------------------------------
+# TestDiscoverClassesFromProvider: supports_durable_execution wiring
+# ---------------------------------------------------------------------------
+class TestDiscoverClassesFromProviderDurableExecution:
+ def test_marks_durable_capable_and_plain_operators(self, tmp_path):
+ class ResumableOperator(FullyImplementedResumableOperator):
+ __module__ = "airflow.providers.test.operators.spark"
+
+ class PlainProviderOperator(PlainOperator):
+ __module__ = "airflow.providers.test.operators.spark"
+
+ provider_yaml = {
+ "package-name": "apache-airflow-providers-test",
+ "name": "Test",
+ "operators": [
+ {
+ "integration-name": "Test",
+ "python-modules":
["airflow.providers.test.operators.spark"],
+ },
+ ],
+ }
+ provider_dir = tmp_path / "test"
+ provider_dir.mkdir()
+ yaml_path = provider_dir / "provider.yaml"
+ yaml_path.write_text(yaml.dump(provider_yaml))
+
+ mod = _make_module(
+ "airflow.providers.test.operators.spark",
+ {
+ "ResumableOperator": ResumableOperator,
+ "PlainProviderOperator": PlainProviderOperator,
+ },
+ )
+
+ with (
+ patch("extract_parameters.PROVIDERS_DIR", tmp_path),
+ patch("extract_parameters.importlib.import_module",
return_value=mod),
+ ):
+ result = discover_classes_from_provider(
+ yaml_path, base_classes={},
resumable_mixin=FakeResumableJobMixin
+ )
+
+ by_name = {r["name"]: r for r in result}
+ assert by_name["ResumableOperator"]["supports_durable_execution"] is
True
+ assert by_name["PlainProviderOperator"]["supports_durable_execution"]
is False
+
+ def test_defaults_to_false_when_mixin_unavailable(self, tmp_path):
+ class ResumableOperator(FullyImplementedResumableOperator):
+ __module__ = "airflow.providers.test.operators.spark"
+
+ provider_yaml = {
+ "package-name": "apache-airflow-providers-test",
+ "name": "Test",
+ "operators": [
+ {
+ "integration-name": "Test",
+ "python-modules":
["airflow.providers.test.operators.spark"],
+ },
+ ],
+ }
+ provider_dir = tmp_path / "test"
+ provider_dir.mkdir()
+ yaml_path = provider_dir / "provider.yaml"
+ yaml_path.write_text(yaml.dump(provider_yaml))
+
+ mod = _make_module(
+ "airflow.providers.test.operators.spark",
+ {"ResumableOperator": ResumableOperator},
+ )
+
+ with (
+ patch("extract_parameters.PROVIDERS_DIR", tmp_path),
+ patch("extract_parameters.importlib.import_module",
return_value=mod),
+ ):
+ result = discover_classes_from_provider(yaml_path, base_classes={})
+
+ assert result[0]["supports_durable_execution"] is False
+
+
# ---------------------------------------------------------------------------
# TestDiscoverClassLevelEntries
# ---------------------------------------------------------------------------
diff --git a/dev/registry/tests/test_registry_contract_models.py
b/dev/registry/tests/test_registry_contract_models.py
index 146a16ff3e6..120c682da78 100644
--- a/dev/registry/tests/test_registry_contract_models.py
+++ b/dev/registry/tests/test_registry_contract_models.py
@@ -22,6 +22,7 @@ import pytest
from pydantic import ValidationError
from registry_contract_models import (
build_openapi_document,
+ validate_modules_catalog,
validate_provider_parameters,
validate_provider_version_metadata,
validate_provider_versions,
@@ -75,6 +76,30 @@ def test_validate_provider_parameters_preserves_mro_alias():
assert "mro_chain" not in class_entry
+def _module_payload(**overrides):
+ payload = {
+ "name": "Example",
+ "type": "operator",
+ "import_path": "airflow.providers.test.mod.Example",
+ "short_description": "Example module.",
+ "docs_url": "https://example.invalid/docs";,
+ "source_url": "https://example.invalid/source";,
+ "category": "test",
+ }
+ payload.update(overrides)
+ return payload
+
+
+def
test_module_contract_accepts_legacy_modules_without_supports_durable_execution():
+ validated = validate_modules_catalog({"modules": [_module_payload()]})
+ assert "supports_durable_execution" not in validated["modules"][0]
+
+
+def test_module_contract_preserves_supports_durable_execution_true():
+ validated = validate_modules_catalog({"modules":
[_module_payload(supports_durable_execution=True)]})
+ assert validated["modules"][0]["supports_durable_execution"] is True
+
+
def
test_validate_version_metadata_accepts_legacy_version_modules_without_ids():
payload = {
"provider_id": "test",
diff --git a/registry/pnpm-workspace.yaml b/registry/pnpm-workspace.yaml
index de82ed55b6e..aae7d571c4b 100644
--- a/registry/pnpm-workspace.yaml
+++ b/registry/pnpm-workspace.yaml
@@ -14,11 +14,13 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-
---
packages:
- '.'
+allowBuilds:
+ '@scarf/scarf': false
+
overrides:
brace-expansion@<1.1.13: '>=1.1.13'
liquidjs@<10.25.0: '>=10.25.0'
diff --git a/registry/src/css/main.css b/registry/src/css/main.css
index a0adbd071c0..2f8f4d4cc6a 100644
--- a/registry/src/css/main.css
+++ b/registry/src/css/main.css
@@ -3252,6 +3252,19 @@ main {
margin-bottom: var(--space-1);
}
+.provider-detail-page .modules .module .content h3 .durable-badge {
+ display: inline-flex;
+ align-items: center;
+ vertical-align: middle;
+ margin-left: var(--space-2);
+ padding: 0.1rem var(--space-2);
+ font-size: var(--text-xs);
+ font-weight: var(--font-medium);
+ border-radius: var(--radius-lg);
+ background: color-mix(in srgb, var(--color-teal-500) 15%, transparent);
+ color: var(--color-teal-400);
+}
+
.provider-detail-page .modules .module .content p {
font-size: var(--text-sm);
color: var(--text-secondary);
diff --git a/registry/src/provider-version.njk
b/registry/src/provider-version.njk
index 51179078bab..49356c0796f 100644
--- a/registry/src/provider-version.njk
+++ b/registry/src/provider-version.njk
@@ -355,7 +355,12 @@ eleventyComputed:
<span>{{ module.type[0] | upper }}</span>
</div>
<div class="content">
- <h3>{{ module.name }}</h3>
+ <h3>
+ {{ module.name }}
+ {% if module.supports_durable_execution %}
+ <span class="durable-badge" title="Reconnects to an
already-running job on retry instead of resubmitting. Requires durable=True
(default) and depends on operator configuration.">Durable</span>
+ {% endif %}
+ </h3>
<p>{{ module.short_description }}</p>
<div class="import-row">
<code class="path">{{ module.import_path }}</code>
