From 66508dba3c27149a11384a5f5aeeebf249f5b5f9 Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Tue, 29 Sep 2026 16:25:27 +0000 Subject: [PATCH 1/2] Move the Dag file processor's shared plumbing into a base class BaseDagFileProcessorProcess holds the parse log forwarding, the request handlers, readiness and cleanup. DagFileProcessorProcess keeps only how the Python parse child starts. No behavior change. --- .../src/airflow/dag_processing/processor.py | 151 ++++++++++-------- 1 file changed, 86 insertions(+), 65 deletions(-) diff --git a/airflow-core/src/airflow/dag_processing/processor.py b/airflow-core/src/airflow/dag_processing/processor.py index 80e6d7161dec2..aa8a1edfa851e 100644 --- a/airflow-core/src/airflow/dag_processing/processor.py +++ b/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,18 +573,17 @@ 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 + logger_filehandle: BinaryIO | None = None parsing_result: DagFileParsingResult | None = None decoder: ClassVar[TypeAdapter[ToManager]] = TypeAdapter[ToManager](ToManager) had_callbacks: bool = False # Track if this process was started with callbacks to prevent stale DAG detection false positives @@ -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, @@ -720,6 +666,8 @@ def wait(self) -> int: def close(self): self.cleanup_sockets_after_kill() + if self.logger_filehandle is None: + return try: self.logger_filehandle.close() except OSError: @@ -728,3 +676,76 @@ 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. + """ + + logger_filehandle: BinaryIO + + _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) From 9024fcd611aee6f66ae51cc0cdf913742215d54c Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Fri, 2 Oct 2026 13:03:49 +0000 Subject: [PATCH 2/2] Keep the parse log handle required on the base Dag file processor The manager gives DagFileProcessorProcess the per-file parse log handle, and the base closes it, as on main. Co-Authored-By: Claude --- airflow-core/src/airflow/dag_processing/processor.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/airflow-core/src/airflow/dag_processing/processor.py b/airflow-core/src/airflow/dag_processing/processor.py index aa8a1edfa851e..d7e107cb0ab1f 100644 --- a/airflow-core/src/airflow/dag_processing/processor.py +++ b/airflow-core/src/airflow/dag_processing/processor.py @@ -583,7 +583,7 @@ class BaseDagFileProcessorProcess(WatchedSubprocess, LoggingMixin): parse request. """ - logger_filehandle: BinaryIO | None = None + logger_filehandle: BinaryIO parsing_result: DagFileParsingResult | None = None decoder: ClassVar[TypeAdapter[ToManager]] = TypeAdapter[ToManager](ToManager) had_callbacks: bool = False # Track if this process was started with callbacks to prevent stale DAG detection false positives @@ -666,8 +666,6 @@ def wait(self) -> int: def close(self): self.cleanup_sockets_after_kill() - if self.logger_filehandle is None: - return try: self.logger_filehandle.close() except OSError: @@ -690,8 +688,6 @@ class DagFileProcessorProcess(BaseDagFileProcessorProcess): in core Airflow. """ - logger_filehandle: BinaryIO - _request_handlers: ClassVar[dict[type[BaseModel], RequestHandler[DagFileProcessorProcess]]] = { **BaseDagFileProcessorProcess._common_request_handlers, }