class_docstr: Runs DAG processing in a separate process using DagFileProcessor.
  
  :param file_path: a Python file containing Airflow DAG definitions
  :param pickle_dags: whether to serialize the DAG objects to the DB
  :param dag_ids: If specified, only look at these DAG ID's
  :param callback_requests: failure callback to execute
class_docstr_tok: ["Runs","DAG","processing","in","a","separate","process","using","DagFileProcessor",".",":","param","file_path",":","a","Python","file","containing","Airflow","DAG","definitions",":","param","pickle_dags",":","whether","to","serialize","the","DAG","objects","to","the","DB",":","param","dag_ids",":","If","specified",",","only","look","at","these","DAG","ID's",":","param","callback_requests",":","failure","callback","to","execute"]
docstr: Process the given file.
  
  :param result_channel: the connection to use for passing back the result
  :param parent_channel: the parent end of the channel to close in the child
  :param file_path: the file to process
  :param pickle_dags: whether to pickle the DAGs found in the file and
      save them to the DB
  :param dag_ids: if specified, only examine DAG ID's that are
      in this list
  :param thread_name: the name to use for the process that is launched
  :param callback_requests: failure callback to execute
  :return: the process that was launched
code: def _run_file_processor(
      result_channel: MultiprocessingConnection,
      parent_channel: MultiprocessingConnection,
      file_path: str,
      pickle_dags: bool,
      dag_ids: list[str] | None,
      thread_name: str,
      dag_directory: str,
      callback_requests: list[CallbackRequest],
  ) -> None:
      """
      Process the given file.
  
      :param result_channel: the connection to use for passing back the result
      :param parent_channel: the parent end of the channel to close in the child
      :param file_path: the file to process
      :param pickle_dags: whether to pickle the DAGs found in the file and
          save them to the DB
      :param dag_ids: if specified, only examine DAG ID's that are
          in this list
      :param thread_name: the name to use for the process that is launched
      :param callback_requests: failure callback to execute
      :return: the process that was launched
      """
      # This helper runs in the newly created process
      log: logging.Logger = logging.getLogger("airflow.processor")
  
      # Since we share all open FDs from the parent, we need to close the parent side of the pipe here in
      # the child, else it won't get closed properly until we exit.
      parent_channel.close()
      del parent_channel
  
      set_context(log, file_path)
      setproctitle(f"airflow scheduler - DagFileProcessor {file_path}")
  
      def _handle_dag_file_processing():
          # Re-configure the ORM engine as there are issues with multiple processes
          settings.configure_orm()
  
          # Change the thread name to differentiate log lines. This is
          # really a separate process, but changing the name of the
          # process doesn't work, so changing the thread name instead.
          threading.current_thread().name = thread_name
  
          log.info(
              "Started process (PID=%s) to work on %s",
              os.getpid(),
              file_path,
          )
          dag_file_processor = DagFileProcessor(
              dag_ids=dag_ids, dag_directory=dag_directory, log=log
          )
          result: tuple[int, int] = dag_file_processor.process_file(
              file_path=file_path,
              pickle_dags=pickle_dags,
              callback_requests=callback_requests,
          )
          result_channel.send(result)
  
      try:
          DAG_PROCESSOR_LOG_TARGET = conf.get_mandatory_value(
              "logging", "DAG_PROCESSOR_LOG_TARGET"
          )
          if DAG_PROCESSOR_LOG_TARGET == "stdout":
              with Stats.timer() as timer:
                  _handle_dag_file_processing()
          else:
              # The following line ensures that stdout goes to the same destination as the logs. If stdout
              # gets sent to logs and logs are sent to stdout, this leads to an infinite loop. This
              # necessitates this conditional based on the value of DAG_PROCESSOR_LOG_TARGET.
              with redirect_stdout(
                  StreamLogWriter(log, logging.INFO)
              ), redirect_stderr(
                  StreamLogWriter(log, logging.WARN)
              ), Stats.timer() as timer:
                  _handle_dag_file_processing()
          log.info(
              "Processing %s took %.3f seconds",
              file_path,
              timer.duration,
          )
      except Exception:
          # Log exceptions through the logging framework.
          log.exception("Got an exception! Propagating...")
          raise
      finally:
          # We re-initialized the ORM within this Process above so we need to
          # tear it down manually here
          settings.dispose_orm()
  
          result_channel.close()
  
imports: from __future__ import annotations
  import importlib
  import logging
  import os
  import signal
  import threading
  import time
  import zipfile
  from contextlib import redirect_stderr, redirect_stdout, suppress
  from datetime import timedelta
  from typing import TYPE_CHECKING, Iterable, Iterator
  from setproctitle import setproctitle
  from sqlalchemy import delete, func, or_, select
  from airflow import settings
  from airflow.api_internal.internal_api_call import internal_api_call
  from airflow.callbacks.callback_requests import DagCallbackRequest, SlaCallbackRequest, TaskCallbackRequest
  from airflow.configuration import conf
  from airflow.exceptions import AirflowException, TaskNotFound
  from airflow.models import SlaMiss, errors
  from airflow.models.dag import DAG, DagModel
  from airflow.models.dagbag import DagBag
  from airflow.models.dagrun import DagRun
  from airflow.models.dagwarning import DagWarning, DagWarningType
  from airflow.models.serialized_dag import SerializedDagModel
  from airflow.models.taskinstance import TaskInstance, TaskInstance
  from airflow.stats import Stats
  from airflow.utils import timezone
  from airflow.utils.email import get_email_address_list, send_email
  from airflow.utils.file import iter_airflow_imports, might_contain_dag
  from airflow.utils.log.logging_mixin import LoggingMixin, StreamLogWriter, set_context
  from airflow.utils.mixins import MultiprocessingStartMethodMixin
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.state import TaskInstanceState
label_desc: Use image node_id 3 for calling the DagFileProcessorProcess obj's underlying member method code with example usage: obj._run_file_processor(result_channel, parent_channel, file_path, pickle_dags, dag_ids, thread_name, dag_directory, callback_requests) without return types