class_docstr: Tfx runner on Airflow.
class_docstr_tok: ["Tfx","runner","on","Airflow","."]
docstr: Deploys given logical pipeline on Airflow.
  
  Args:
    tfx_pipeline: Logical pipeline containing pipeline args and components.
  
  Returns:
    An Airflow DAG.
docstr_tok: ["Deploys","given","logical","pipeline","on","Airflow",".","Args",":","tfx_pipeline",":","Logical","pipeline","containing","pipeline","args","and","components",".","Returns",":","An","Airflow","DAG","."]
code: def run(self, tfx_pipeline: pipeline.Pipeline):
      """Deploys given logical pipeline on Airflow.
  
      Args:
        tfx_pipeline: Logical pipeline containing pipeline args and components.
  
      Returns:
        An Airflow DAG.
      """
  
      # Merge airflow-specific configs with pipeline args
      airflow_dag = models.DAG(
          dag_id=tfx_pipeline.pipeline_info.pipeline_name,
          **(
              typing.cast(
                  AirflowPipelineConfig, self._config
              ).airflow_dag_config
          ),
      )
      if "tmp_dir" not in tfx_pipeline.additional_pipeline_args:
          tmp_dir = os.path.join(
              tfx_pipeline.pipeline_info.pipeline_root, ".temp", ""
          )
          tfx_pipeline.additional_pipeline_args["tmp_dir"] = tmp_dir
  
      component_impl_map = {}
      for tfx_component in tfx_pipeline.components:
          # TODO(b/187122662): Pass through pip dependencies as a first-class
          # component flag.
          if isinstance(tfx_component, base_component.BaseComponent):
              tfx_component._resolve_pip_dependencies(  # pylint: disable=protected-access
                  tfx_pipeline.pipeline_info.pipeline_root
              )
  
          tfx_component = self._replace_runtime_params(tfx_component)
  
          (
              component_launcher_class,
              component_config,
          ) = config_utils.find_component_launch_info(
              self._config, tfx_component
          )
          current_airflow_component = airflow_component.AirflowComponent(
              parent_dag=airflow_dag,
              component=tfx_component,
              component_launcher_class=component_launcher_class,
              pipeline_info=tfx_pipeline.pipeline_info,
              enable_cache=tfx_pipeline.enable_cache,
              metadata_connection_config=tfx_pipeline.metadata_connection_config,
              beam_pipeline_args=tfx_pipeline.beam_pipeline_args,
              additional_pipeline_args=tfx_pipeline.additional_pipeline_args,
              component_config=component_config,
          )
          component_impl_map[tfx_component] = current_airflow_component
          for upstream_node in tfx_component.upstream_nodes:
              assert upstream_node in component_impl_map, (
                  "Components is not in " "topological order"
              )
              current_airflow_component.set_upstream(
                  component_impl_map[upstream_node]
              )
  
      return airflow_dag
  
imports: import os
  import typing
  from typing import Any, Dict, Optional, Union
  import warnings
  from airflow import models
  from tfx.dsl.components.base import base_component
  from tfx.orchestration import pipeline
  from tfx.orchestration import tfx_runner
  from tfx.orchestration.airflow import airflow_component
  from tfx.orchestration.config import config_utils
  from tfx.orchestration.config import pipeline_config
  from tfx.orchestration.data_types import RuntimeParameter
  from tfx.utils.json_utils import json
label_desc: Use image node_id 2 for calling the AirflowDagRunner obj's underlying member method code with example usage: obj.run(tfx_pipeline) and returns: airflow_dag