code: def external_pipeline_artifact_query(
      artifact_type: Type[artifact.Artifact],
      *,
      owner: str,
      pipeline_name: str,
      producer_component_id: str,
      output_key: str,
      pipeline_run_id: str = "",
  ) -> channel.ExternalPipelineChannel:
      """Helper function to construct a query to get artifacts from an external pipeline.
  
      Args:
        artifact_type: Subclass of Artifact for this channel.
        owner: Onwer of the pipeline.
        pipeline_name: Name of the pipeline the artifacts belong to.
        producer_component_id: Id of the component produces the artifacts.
        output_key: The output key when producer component produces the artifacts in
          this Channel.
        pipeline_run_id: (Optional) Pipeline run id the artifacts belong to.
  
      Returns:
        channel.ExternalPipelineChannel instance.
  
      Raises:
        ValueError, if owner or pipeline_name is missing.
      """
      if not owner or not pipeline_name:
          raise ValueError("owner or pipeline_name is missing.")
  
      return channel.ExternalPipelineChannel(
          artifact_type=artifact_type,
          owner=owner,
          pipeline_name=pipeline_name,
          producer_component_id=producer_component_id,
          output_key=output_key,
          pipeline_run_id=pipeline_run_id,
      )
  
imports: import typing
  from typing import Callable, Dict, Iterable, Iterator, List, Optional, Sequence, Set, Type, cast
  from tfx.dsl.placeholder import placeholder
  from tfx.proto.orchestration import placeholder_pb2
  from tfx.types import artifact
  from tfx.types import channel
label_desc: Use image node_id 6 for calling a global function with example usage: external_pipeline_artifact_query(artifact_type) and returns: channel