bbovenzi commented on code in PR #74278:
URL: https://github.com/apache/airflow/pull/74278#discussion_r4198064776
##########
airflow-core/src/airflow/plugins_manager.py:
##########
@@ -158,40 +160,153 @@ def __register_plugins(plugin_instances:
list[AirflowPlugin], errors: dict[str,
return plugins, import_errors
-_DAG_APPLIES_TO_CRITERIA = frozenset({"dag_tags", "dag_ids"})
-_TASK_APPLIES_TO_CRITERIA = frozenset({"task_ids", "operators",
"operator_names"})
-_APPLIES_TO_CRITERIA = _DAG_APPLIES_TO_CRITERIA | _TASK_APPLIES_TO_CRITERIA
-
-# Which `applies_to` criteria each destination can resolve a record for. A
destination
-# missing from this mapping cannot evaluate any criterion. Kept in sync with
the table in
+# The records each destination can resolve, and therefore the path roots it
can evaluate. A
+# destination missing from this mapping can evaluate nothing. Kept in sync
with the table in
# docs/administration-and-deployment/plugins.rst.
-_EVALUABLE_CRITERIA_BY_DESTINATION: dict[str, frozenset[str]] = {
- "dag": _DAG_APPLIES_TO_CRITERIA,
- "dag_run": _DAG_APPLIES_TO_CRITERIA,
- "dag_overview": _DAG_APPLIES_TO_CRITERIA,
- "task": _APPLIES_TO_CRITERIA,
- "task_overview": _APPLIES_TO_CRITERIA,
- "task_instance": _APPLIES_TO_CRITERIA,
+_APPLIES_TO_ROOTS: dict[str, frozenset[str]] = {
+ "dag": frozenset({"dag"}),
+ "dag_overview": frozenset({"dag"}),
+ "dag_run": frozenset({"dag", "dag_run"}),
+ "task": frozenset({"dag", "task"}),
+ "task_overview": frozenset({"dag", "task"}),
+ "task_instance": frozenset({"dag", "dag_run", "task", "task_instance"}),
"nav": frozenset(),
"base": frozenset(),
"dashboard": frozenset(),
"asset": frozenset(),
}
+# Which record an unqualified path is rooted at -- the entity the destination
is about.
+_APPLIES_TO_ENTITY_ROOT: dict[str, str] = {
+ "dag": "dag",
+ "dag_overview": "dag",
+ "dag_run": "dag_run",
+ "task": "task",
+ "task_overview": "task",
+ "task_instance": "task_instance",
+}
+
+_APPLIES_TO_ROOT_NAMES = frozenset({"dag", "dag_run", "task", "task_instance"})
+
+
+def _applies_to_path_root(path: str, destination: str) -> str | None:
+ """
+ Return the record a path is rooted at, or ``None`` if the destination has
no entity.
+
+ A path may name a related record as its first segment; otherwise it is
rooted at the
+ entity the destination is about.
+ """
+ head, _, rest = path.partition(".")
+ if head in _APPLIES_TO_ROOT_NAMES and rest:
+ return head
+ return _APPLIES_TO_ENTITY_ROOT.get(destination)
+
+
+# Sentinel for an annotation that does not describe what it contains, so a
path cannot be
+# checked past it.
+_OPAQUE = object()
+
+
+@cache
+def _applies_to_root_models() -> dict[str, Any]:
+ """
+ Return the response model backing each path root, for validating paths at
plugin load.
+
+ Imported lazily because ``datamodels.plugins`` imports this module; a
module-level import
+ would be circular. Only ``_get_ui_plugins`` reaches this, so components
that load plugins
+ without serving the UI never pay for it.
+
+ These are the models the UI actually fetches for the ``applies_to``
context -- keep them in
+ step with ``AppliesToContext`` in ``src/utils/pluginAppliesTo.ts``.
+ """
+ from airflow.api_fastapi.core_api.datamodels.dag_run import DAGRunResponse
+ from airflow.api_fastapi.core_api.datamodels.dags import DAGResponse
+ from airflow.api_fastapi.core_api.datamodels.task_instances import
TaskInstanceResponse
+ from airflow.api_fastapi.core_api.datamodels.tasks import TaskResponse
+
+ return {
+ "dag": DAGResponse,
+ "dag_run": DAGRunResponse,
+ "task": TaskResponse,
+ "task_instance": TaskInstanceResponse,
+ }
+
+
+def _unwrap_applies_to_annotation(annotation: Any) -> Any:
+ """
+ Reduce a field annotation to the type a further path segment reads through.
+
+ ``Annotated`` and ``X | None`` wrappers are stripped, and a list is
stepped into, because
+ traversing one fans out across its elements. Returns ``_OPAQUE`` for a
union of several
+ real types, whose fields depend on which member a record actually holds.
+ """
+ while True:
+ origin = get_origin(annotation)
+ if origin is Annotated:
+ annotation = get_args(annotation)[0]
+ elif origin in (Union, types.UnionType):
+ members = [arg for arg in get_args(annotation) if arg is not
type(None)]
+ if len(members) != 1:
+ return _OPAQUE
+ annotation = members[0]
+ elif origin in (list, set, frozenset, tuple):
+ args = get_args(annotation)
+ if not args:
+ return _OPAQUE
+ annotation = args[0]
+ else:
+ return annotation
+
+
+def _describe_applies_to_path_error(path: str, root: str) -> str | None:
+ """
+ Return why ``path`` names no field on its root record, or ``None`` if it
is not knowably wrong.
+
+ The walk stops -- accepting whatever follows -- at a field the models do
not describe the
+ contents of, such as the bare ``dict`` behind ``class_ref`` or a Dag Run's
``conf``. So this
+ catches a misspelling of a modelled field, not every bad path.
+ """
+ segments = path.split(".")
+ # A qualified path's first segment names the record, which `root` has
already resolved.
+ if segments[0] == root and len(segments) > 1:
+ segments = segments[1:]
+
+ current: Any = _applies_to_root_models()[root]
+ for segment in segments:
+ if current is _OPAQUE or current is Any:
+ return None
+ base = get_origin(current) or current
+ if isinstance(base, type) and issubclass(base, dict):
+ return None
+
+ fields = getattr(current, "model_fields", None)
Review Comment:
Updated to use the fields available on FastAPI's routes.
--
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]
