jason810496 commented on code in PR #71151:
URL: https://github.com/apache/airflow/pull/71151#discussion_r3802217039
##########
contributing-docs/30_new_language_sdk.rst:
##########
@@ -491,6 +496,32 @@ native Dags they are *not applicable* (``n/a``) rather
than unsupported.
The SDK exposes an object-storage API (an ``ObjectStoragePath``
equivalent) usable from
native Dag code.
+Compatibility matrix
+~~~~~~~~~~~~~~~~~~~~
+
+The dimensions above are prose; each SDK also declares them *machine-readably*
in one
+hand-authored ``<sdk>/capabilities.yaml``, from which a prek hook generates
every table:
+
+.. code-block:: text
+
+ java-sdk/capabilities.yaml <- the only file you edit
+ |
+ | hook: update-java-sdk-readme-matrix
+ |
+ +--> java-sdk/README.md (contributor-facing)
+ +--> java-sdk/sdk/module.md (Dokka -> the published API reference)
+ +--> airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst
+ (consolidated cross-SDK matrix — TODO, not generated yet)
Review Comment:
https://github.com/apache/airflow/pull/71757 will wire the matrix for all
the SDKs. I remove the TODO here for now.
##########
java-sdk/sdk/dokka/matrix.css:
##########
@@ -0,0 +1,42 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/* Dokka sizes content-table cells with `min-width: 40px` against ~31px of
horizontal padding, so the
+ narrow columns of the Language SDK compatibility matrix collapse until
their headers and values
+ break mid-word ("Supported", "MUST †").
+
+ These selectors necessarily apply to every content table in the API
reference, because a table
+ generated from Markdown carries no class of its own. They are therefore
deliberately written to
+ only *raise a floor*, never to forbid wrapping: a header stays on one line
(headers are short in
+ any table), and the narrow value columns get a minimum width instead of
`nowrap`. An unrelated
+ table with long prose in those columns still wraps as before rather than
overflowing. */
Review Comment:
Agreed, I trimmed the comments in
https://github.com/apache/airflow/pull/71151/changes/aa3b1b3ac48ef9e4f7bdc9f8d75d4a03f3533fe8#diff-f58bd21e679ab8560bf95a8473dc5a3b428d54f841f3475f69e05f05795a89a7
, thanks.
##########
scripts/ci/prek/lang_sdk_compat_matrix.py:
##########
@@ -0,0 +1,294 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Shared helpers for the Language SDK compatibility matrix.
+
+Every Language SDK (Go, Java, TypeScript) declares what it supports in a
hand-authored
+``<sdk>/capabilities.yaml`` at the root of its own tree. This module owns the
*schema* of that
+file, the *registry* of SDKs, and :func:`render_markdown_table`, which renders
the per-SDK Markdown
+table that each SDK's own prek hook embeds in its docs.
+
+Because the manifest is hand-authored, :func:`validate_capabilities` is the
only thing standing
+between a typo and a wrong published table, so it rejects unknown keys as well
as missing ones.
+
+The normative meaning of each dimension lives in
``contributing-docs/30_new_language_sdk.rst``
+(the "Conformance" section). Keep the dimensions below in sync with that
document.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import NamedTuple, TypedDict
+
+import yaml
+from common_prek_utils import AIRFLOW_ROOT_PATH
+
+# Markers wrapping the generated tables. insert_documentation() keeps these
lines and rewrites
+# everything between them, so the same constants are reused by every SDK's
README hook.
+README_MATRIX_HEADER = "<!-- BEGIN AUTO-GENERATED LANG-SDK COMPAT MATRIX -->"
+README_MATRIX_FOOTER = "<!-- END AUTO-GENERATED LANG-SDK COMPAT MATRIX -->"
+
+SUPPORTED_MARK = "✓"
+UNSUPPORTED_MARK = "✗"
+NA_MARK = "n/a" # a gated native-Dag capability while native-dag-authoring is
unsupported
+NO_VERSION_MARK = "–" # "Since" placeholder for a dimension that is not
supported
+
+# The umbrella capability that gates the conditional native-Dag capabilities:
when an SDK does
+# not support it, every gated native-Dag capability is "not applicable" rather
than unsupported.
+NATIVE_DAG_GATE = "native-dag-authoring"
+
+# TaskInstance states a subprocess can emit, in display order, with their
conformance tier.
+# Scheduler-owned states (queued, scheduled, running, restarting,
upstream_failed) are never
+# emitted by an SDK runtime and are deliberately excluded.
+STATE_DIMENSIONS: list[tuple[str, str]] = [
+ ("success", "MUST"),
+ ("failed", "MUST"),
+ ("up_for_retry", "MUST"),
+ ("skipped", "SHOULD"),
+ ("deferred", "MAY"),
+ ("up_for_reschedule", "MAY"),
+ ("awaiting_input", "MAY"),
+ ("removed", "MAY"),
+]
+
+
+class Capability(NamedTuple):
+ name: str
+ tier: str
+ group: str # "runtime" or "native"
+ gated: bool # renders n/a (not ✗) when NATIVE_DAG_GATE is unsupported
+
+
+# Capability flags, in display order. Runtime capabilities describe what a
task body can do while
+# it runs (in either a mixed-lang or native Dag); native-Dag capabilities
describe authoring a
+# whole Dag in the target language and are gated by NATIVE_DAG_GATE (except
the gate itself).
+CAPABILITY_DIMENSIONS: list[Capability] = [
+ Capability("mixed-lang-stub-target", "MUST", "runtime", False),
+ Capability("task-logging", "MUST", "runtime", False),
+ Capability("xcom-read-write", "MUST", "runtime", False),
+ Capability("connection-read", "MUST", "runtime", False),
+ Capability("variable-read-write", "MUST", "runtime", False),
+ Capability("self-contained-bundle", "MUST", "runtime", False),
+ Capability("retry-policy", "MAY", "runtime", False),
+ Capability("task-state-store", "MAY", "runtime", False),
+ Capability("asset-state-store", "MAY", "runtime", False),
+ Capability("asset-event-emit", "MAY", "runtime", False),
+ Capability("asset-event-read", "MAY", "runtime", False),
+ Capability(NATIVE_DAG_GATE, "SHOULD", "native", False),
+ Capability("task-args", "MUST", "native", True),
+ Capability("dag-params", "MUST", "native", True),
+ Capability("taskflow-dependencies", "MUST", "native", True),
+ Capability("branching", "SHOULD", "native", True),
+ Capability("dag-test", "SHOULD", "native", True),
+ Capability("task-group", "MAY", "native", True),
+ Capability("dynamic-task-mapping", "MAY", "native", True),
+ Capability("asset-inlets-outlets", "MAY", "native", True),
+ Capability("asset-scheduling", "MAY", "native", True),
+ Capability("object-store", "MAY", "native", True),
+]
+
+CAPABILITY_NAMES = {cap.name for cap in CAPABILITY_DIMENSIONS}
+
+GROUP_LABELS = {"runtime": "Runtime capabilities", "native": "Native-Dag
authoring"}
+STATES_GROUP_LABEL = "TaskInstance states"
+
+LEGEND = (
+ f"Marks: {SUPPORTED_MARK} supported · {UNSUPPORTED_MARK} not supported · "
+ f"{NA_MARK} not applicable. A tier marked † applies only when
`{NATIVE_DAG_GATE}` is supported."
+)
+
+
+class LangSdk(TypedDict):
+ id: str
+ capabilities_yaml: Path
+ readme: Path
+
+
+# The registry of Language SDKs and where each one's manifest and README live.
Only the Java SDK
+# declares one so far; the Go and TypeScript entries record where theirs go
when those runtimes
+# declare their capabilities. Because of that, `capabilities_yaml` is a
declared location and not a
+# promise the file exists — a consumer walking the whole registry must check
`.exists()` before
+# calling load_capabilities().
+LANG_SDKS: list[LangSdk] = [
+ {
+ "id": "go",
+ "capabilities_yaml": AIRFLOW_ROOT_PATH / "go-sdk" /
"capabilities.yaml",
+ "readme": AIRFLOW_ROOT_PATH / "go-sdk" / "README.md",
+ },
+ {
+ "id": "java",
+ "capabilities_yaml": AIRFLOW_ROOT_PATH / "java-sdk" /
"capabilities.yaml",
+ "readme": AIRFLOW_ROOT_PATH / "java-sdk" / "README.md",
+ },
+ {
+ "id": "ts",
+ "capabilities_yaml": AIRFLOW_ROOT_PATH / "ts-sdk" /
"capabilities.yaml",
+ "readme": AIRFLOW_ROOT_PATH / "ts-sdk" / "README.md",
+ },
+]
+
+VALID_SDK_IDS = {sdk["id"] for sdk in LANG_SDKS}
+
+
+class DimensionEntry(TypedDict, total=False):
+ supported: bool
+ since: str | None
+ note: str
+
+
+class CapabilitiesDoc(TypedDict):
+ sdk: str
+ supervisor_schema_version: str
+ min_airflow_version: str
+ states: dict[str, DimensionEntry]
+ capabilities: dict[str, DimensionEntry]
+
+
+class CapabilitiesError(ValueError):
+ """Raised when a capabilities.yaml file does not match the expected
schema."""
+
+
+def load_capabilities(path: Path, *, expected_sdk: str | None = None) ->
CapabilitiesDoc:
+ """Load and validate a ``capabilities.yaml`` file.
+
+ ``expected_sdk`` binds the file to the SDK it belongs to: passing it makes
a manifest whose
+ ``sdk`` field disagrees with the file's own SDK (e.g.
``go-sdk/capabilities.yaml`` declaring
+ ``sdk: java``) a validation error instead of silently rendering in the
wrong column.
+ """
+ doc = yaml.safe_load(path.read_text())
+ validate_capabilities(doc, source=str(path), expected_sdk=expected_sdk)
+ return doc
+
+
+def validate_capabilities(doc: object, *, source: str, expected_sdk: str |
None = None) -> None:
+ """Validate a decoded capabilities document, raising
:class:`CapabilitiesError` on any issue.
+
+ When ``expected_sdk`` is given, the document's ``sdk`` field must equal it.
+ """
+ if not isinstance(doc, dict):
+ raise CapabilitiesError(f"{source}: top-level value must be a mapping")
+ required = {"sdk", "supervisor_schema_version", "min_airflow_version",
"states", "capabilities"}
+ missing = required - doc.keys()
+ if missing:
+ raise CapabilitiesError(f"{source}: missing required keys: {',
'.join(sorted(missing))}")
+ unknown = doc.keys() - required
+ if unknown:
+ raise CapabilitiesError(f"{source}: unknown top-level keys: {',
'.join(sorted(unknown))}")
+ if doc["sdk"] not in VALID_SDK_IDS:
+ raise CapabilitiesError(
+ f"{source}: unknown sdk {doc['sdk']!r}; expected one of
{sorted(VALID_SDK_IDS)}"
+ )
+ if expected_sdk is not None and doc["sdk"] != expected_sdk:
+ raise CapabilitiesError(f"{source}: sdk is {doc['sdk']!r} but this
file belongs to {expected_sdk!r}")
+ for field in ("supervisor_schema_version", "min_airflow_version"):
+ if not isinstance(doc[field], str):
+ raise CapabilitiesError(f"{source}: {field} must be a string")
+ _validate_entries(
+ doc["states"], expected={state for state, _ in STATE_DIMENSIONS},
kind="states", source=source
+ )
+ _validate_entries(doc["capabilities"], expected=CAPABILITY_NAMES,
kind="capabilities", source=source)
+
+
+def _validate_entries(entries: object, *, expected: set[str], kind: str,
source: str) -> None:
+ if not isinstance(entries, dict):
+ raise CapabilitiesError(f"{source}: {kind!r} must be a mapping")
+ actual = set(entries.keys())
+ if actual != expected:
+ missing = expected - actual
+ unknown = actual - expected
+ problems = []
+ if missing:
+ problems.append(f"missing {sorted(missing)}")
+ if unknown:
+ problems.append(f"unknown {sorted(unknown)}")
+ raise CapabilitiesError(f"{source}: {kind} keys mismatch: {';
'.join(problems)}")
+ for name, entry in entries.items():
+ if not isinstance(entry, dict) or not
isinstance(entry.get("supported"), bool):
+ raise CapabilitiesError(f"{source}: {kind}.{name} must be a
mapping with a boolean 'supported'")
+ # A misspelled optional key would otherwise be dropped silently and
render as a blank cell.
+ unknown_fields = entry.keys() - {"supported", "since", "note"}
+ if unknown_fields:
+ raise CapabilitiesError(
+ f"{source}: {kind}.{name} has unknown keys: {',
'.join(sorted(unknown_fields))}"
+ )
+ if not isinstance(entry.get("since", None), (str, type(None))):
+ raise CapabilitiesError(f"{source}: {kind}.{name}.since must be a
string or null")
+ if not entry["supported"] and entry.get("since") is not None:
Review Comment:
No harm to have a strict validation. I addressed in
https://github.com/apache/airflow/pull/71151/changes/aa3b1b3ac48ef9e4f7bdc9f8d75d4a03f3533fe8,
thanks.
--
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]
