code: def record_pipeline(
      output_dir: str,
      metadata_db_uri: Optional[str] = None,
      host: Optional[str] = None,
      port: Optional[int] = None,
      pipeline_name: Optional[str] = None,
      run_id: Optional[str] = None,
  ) -> None:
      """Record pipeline run with run_id to output_dir.
  
      For the beam pipeline, metadata_db_uri is required. For KFP pipeline,
      host and port should be specified. If run_id is not specified, then
      pipeline_name ought to be specified in order to fetch the latest execution
      for the specified pipeline.
  
      Args:
        output_dir: Directory path where the pipeline outputs should be recorded.
        metadata_db_uri: Uri to metadata db.
        host: Hostname of the metadata grpc server
        port: Port number of the metadata grpc server.
        pipeline_name: Pipeline name, which is required if run_id isn't specified.
        run_id: Pipeline execution run_id.
  
      Raises:
        ValueError: In cases of invalid arguments:
          - metadata_db_uri is None or host and/or port is None.
          - run_id is None and pipeline_name is None.
        FileNotFoundError: if the source artifact uri does not already exist.
      """
      if host is not None and port is not None:
          metadata_config = (
              metadata_store_pb2.MetadataStoreClientConfig(
                  host=host, port=port
              )
          )
      elif metadata_db_uri is not None:
          metadata_config = metadata.sqlite_metadata_connection_config(
              metadata_db_uri
          )
      else:
          raise ValueError(
              "For KFP, host and port are required. "
              "For beam pipeline, metadata_db_uri is required."
          )
  
      with metadata.Metadata(metadata_config) as metadata_connection:
          if run_id is None:
              if pipeline_name is None:
                  raise ValueError(
                      "If the run_id is not specified,"
                      " pipeline_name should be specified"
                  )
              # fetch executions of the most recently updated execution context.
              executions = get_latest_executions(
                  metadata_connection, pipeline_name
              )
          else:
              execution_dict = _get_execution_dict(metadata_connection)
              if run_id in execution_dict:
                  executions = execution_dict[run_id]
              else:
                  raise ValueError(
                      "run_id {} is not recorded in the MLMD metadata".format(
                          run_id
                      )
                  )
  
          for src_uri, dest_uri in _get_paths(
              metadata_connection, executions, output_dir
          ):
              io_utils.copy_dir(src_uri, dest_uri)
          logging.info("Pipeline Recorded at %s", output_dir)
  
file_path: tfx/tfx/experimental/pipeline_testing/pipeline_recorder_utils.py
imports: import collections
  import os
  from typing import Iterable, List, Mapping, Optional, Tuple
  from absl import logging
  from tfx.dsl.compiler import constants
  from tfx.orchestration import metadata
  from tfx.utils import io_utils
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 5 for calling a global function with example usage: record_pipeline(output_dir, metadata_db_uri, host, port, pipeline_name, run_id) without return types