Move the Dag file processor's shared plumbing into a base class

airflow-core/src/airflow/dag_processing/processor.py

@@ -92,7 +92,7 @@
     from airflow.sdk.definitions.context import Context
     from airflow.sdk.definitions.dag import DAG
     from airflow.sdk.definitions.mappedoperator import MappedOperator
-    from airflow.sdk.execution_time.supervisor import RequestHandler, RequestResult
+    from airflow.sdk.execution_time.supervisor import RequestHandler, RequestResult, ResponseSent
     from airflow.typing_compat import Self
 
 
@@ -573,15 +573,14 @@ def in_process_api_server() -> InProcessExecutionAPI:
 
 
 @attrs.define(kw_only=True)
-class DagFileProcessorProcess(WatchedSubprocess, LoggingMixin):
+class BaseDagFileProcessorProcess(WatchedSubprocess, LoggingMixin):
     """
-    Parses dags with Task SDK API.
-
-    This class provides a wrapper and management around a subprocess to parse a specific DAG file.
+    Parse one Dag file in a child process for the Dag processor manager.
 
-    Since DAGs are written with the Task SDK, we need to parse them in a task SDK process such that
-    we can use the Task SDK definitions when serializing. This prevents potential conflicts with classes
-    in core Airflow.
+    The child's output goes to the file's parse log, and its requests are answered with
+    :attr:`client`. The parse is done once the child has exited and all its sockets are closed;
+    :attr:`parsing_result` then holds what it sent. Subclasses start the child and send it the
+    parse request.
     """
 
     logger_filehandle: BinaryIO
@@ -595,60 +594,6 @@ class DagFileProcessorProcess(WatchedSubprocess, LoggingMixin):
     bundle_name: str
     dag_file_rel_path: str
 
-    @classmethod
-    def start(  # type: ignore[override]
-        cls,
-        *,
-        path: str | os.PathLike[str],
-        bundle_path: Path,
-        bundle_name: str,
-        dag_file_rel_path: str,
-        callbacks: list[CallbackRequest],
-        target: Callable[[], None] = _parse_file_entrypoint,
-        client: Client,
-        **kwargs,
-    ) -> Self:
-        logger = kwargs["logger"]
-
-        # Parsing DAG files runs user code that can trigger macOS-unsafe ObjC
-        # initialization (secret backends, connection/variable lookups, HTTP
-        # clients). Fork+exec a clean interpreter there. Tests override `target`
-        # with a stub to exercise the base infrastructure; keep bare fork for those.
-        use_exec = target is _parse_file_entrypoint and supervisor._should_use_exec()
-
-        # Pre-importing only helps the bare-fork child (it inherits the imports via
-        # copy-on-write). An exec'd child re-imports from scratch, so skip it there
-        # to avoid leaking user modules into the long-lived processor manager.
-        if not use_exec:
-            _pre_import_airflow_modules(os.fspath(path), logger)
-
-        proc: Self = super().start(
-            target=target,
-            client=client,
-            bundle_name=bundle_name,
-            dag_file_rel_path=dag_file_rel_path,
-            use_exec=use_exec,
-            **kwargs,
-        )
-        proc.had_callbacks = bool(callbacks)  # Track if this process had callbacks
-        proc._on_child_started(callbacks, path, bundle_path, bundle_name)
-        return proc
-
-    def _on_child_started(
-        self,
-        callbacks: list[CallbackRequest],
-        path: str | os.PathLike[str],
-        bundle_path: Path,
-        bundle_name: str,
-    ) -> None:
-        msg = DagFileParseRequest(
-            file=os.fspath(path),
-            bundle_path=bundle_path,
-            bundle_name=bundle_name,
-            callback_requests=callbacks,
-        )
-        self.send_msg(msg, request_id=0)
-
     def _get_target_loggers(self) -> tuple[FilteringBoundLogger, ...]:
         base = super()._get_target_loggers()
         if not self.subprocess_logs_to_stdout:
@@ -674,11 +619,12 @@ def _create_log_forwarder(
 
     def _handle_parsing_result(
         self, msg: DagFileParsingResult, log: FilteringBoundLogger, req_id: int
-    ) -> RequestResult:
+    ) -> RequestResult | ResponseSent:
         self.parsing_result = msg
         return None, {}
 
-    _request_handlers: ClassVar[dict[type[BaseModel], RequestHandler[DagFileProcessorProcess]]] = {
+    # Each subclass builds its own ``_request_handlers``, typed with itself, from these.
+    _common_request_handlers: ClassVar[dict[type[BaseModel], RequestHandler[BaseDagFileProcessorProcess]]] = {
         **WatchedSubprocess._get_shared_request_handlers(
             DeleteVariable,
             GetConnection,
@@ -728,3 +674,74 @@ def close(self):
                 self.dag_file_rel_path,
                 exc_info=True,
             )
+
+
+@attrs.define(kw_only=True)
+class DagFileProcessorProcess(BaseDagFileProcessorProcess):
+    """
+    Parses dags with Task SDK API.
+
+    This class provides a wrapper and management around a subprocess to parse a specific DAG file.
+
+    Since DAGs are written with the Task SDK, we need to parse them in a task SDK process such that
+    we can use the Task SDK definitions when serializing. This prevents potential conflicts with classes
+    in core Airflow.
+    """
+
+    _request_handlers: ClassVar[dict[type[BaseModel], RequestHandler[DagFileProcessorProcess]]] = {
+        **BaseDagFileProcessorProcess._common_request_handlers,
+    }
+
+    @classmethod
+    def start(  # type: ignore[override]
+        cls,
+        *,
+        path: str | os.PathLike[str],
+        bundle_path: Path,
+        bundle_name: str,
+        dag_file_rel_path: str,
+        callbacks: list[CallbackRequest],
+        target: Callable[[], None] = _parse_file_entrypoint,
+        client: Client,
+        **kwargs,
+    ) -> Self:
+        logger = kwargs["logger"]
+
+        # Parsing DAG files runs user code that can trigger macOS-unsafe ObjC
+        # initialization (secret backends, connection/variable lookups, HTTP
+        # clients). Fork+exec a clean interpreter there. Tests override `target`
+        # with a stub to exercise the base infrastructure; keep bare fork for those.
+        use_exec = target is _parse_file_entrypoint and supervisor._should_use_exec()
+
+        # Pre-importing only helps the bare-fork child (it inherits the imports via
+        # copy-on-write). An exec'd child re-imports from scratch, so skip it there
+        # to avoid leaking user modules into the long-lived processor manager.
+        if not use_exec:
+            _pre_import_airflow_modules(os.fspath(path), logger)
+
+        proc: Self = super().start(
+            target=target,
+            client=client,
+            bundle_name=bundle_name,
+            dag_file_rel_path=dag_file_rel_path,
+            use_exec=use_exec,
+            **kwargs,
+        )
+        proc.had_callbacks = bool(callbacks)  # Track if this process had callbacks
+        proc._on_child_started(callbacks, path, bundle_path, bundle_name)
+        return proc
+
+    def _on_child_started(
+        self,
+        callbacks: list[CallbackRequest],
+        path: str | os.PathLike[str],
+        bundle_path: Path,
+        bundle_name: str,
+    ) -> None:
+        msg = DagFileParseRequest(
+            file=os.fspath(path),
+            bundle_path=bundle_path,
+            bundle_name=bundle_name,
+            callback_requests=callbacks,
+        )
+        self.send_msg(msg, request_id=0)