uranusjr commented on code in PR #75405:
URL: https://github.com/apache/airflow/pull/75405#discussion_r4465769245
##########
task-sdk/src/airflow/sdk/importers/yaml_importer/builder.py:
##########
@@ -105,11 +108,64 @@ def build(self, doc: DagDocument) -> DAG:
for task in doc.tasks:
for upstream in sorted(task.get_dependency_ids()):
self._operators[upstream] >> self._operators[task.id_]
+ self._validate_stub_handlers(doc)
dag.bundle_name = self._bundle.name
dag.fileloc = repr(self._definition)
dag.relative_fileloc =
self._definition.get_relative_loc(self._bundle.path)
return dag
+ def _validate_stub_handlers(self, doc: DagDocument) -> None:
+ """
+ Check each ``run:`` task's arguments against its handler.
+
+ This implements the interface defined in AIP-108 and Language SDK
+ ADR-0011. The routed coordinator is queried to obtain the handler's
+ signature and compared against the ``run:`` arguments. Mismatches are
+ collected and raised together as a :class:`YamlDagBuildError`,
+ eventually recorded as an import error.
+
+ Validation is skipped when no declaration is available (no handler
+ bundle or an unregistered handler), or resolving one fails (an
+ uninitializable bundle, an ImportError in it, or a malformed handler).
+
+ This never fails parsing, so building a Dag does not depend on handler
+ availability. A real problem surfaces when the task runs.
+ """
+ if not (stub_tasks := [task for task in doc.tasks if
self._operators[task.id_].is_stub]):
+ return
+ if problems := "; ".join(self._iter_stub_handler_collect_problems(doc,
stub_tasks)):
+ raise YamlDagBuildError(problems)
+
+ def _iter_stub_handler_collect_problems(self, doc: DagDocument,
stub_tasks: list[Task]) -> Iterator[str]:
+ from airflow.sdk.execution_time.coordinator import
get_coordinator_manager
+
+ manager = get_coordinator_manager()
+ for task in stub_tasks:
+ coordinator = manager.for_queue(self._operators[task.id_].queue)
+ try:
+ declaration = coordinator.parse_task_handler(
+ dag_id=doc.dag_id,
+ task_id=task.id_,
+ bundle=self._bundle,
+ )
+ except Exception:
Review Comment:
Block added (I decided to use a dedicated exception type instead of
ValueError)
--
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]
