class_docstr: A channel parameter that forms part of a ComponentSpec.
  
  This type of parameter should be specified in the INPUTS and OUTPUTS dict
  fields of a ComponentSpec:
  
  class MyCustomComponentSpec(ComponentSpec):
    # ...
    INPUTS = {
        'input_examples': ChannelParameter(type=standard_artifacts.Examples),
    }
    OUTPUTS = {
        'output_examples': ChannelParameter(type=standard_artifacts.Examples),
    }
    # ...
docstr: ChannelParameter constructor.
  
  Note the distinction between `optional` and `allow_empty`.
  
  An illustrative example: a component may make it optional to provide a
  specific channel C during pipeline definition time (i.e. set `optional` to
  True). However, it may require that if the channel C is provided during
  pipeline definition, during orchestration, the channel must have artifacts
  before the component is triggered (i.e. set `allow_empty` to False).
  
  Args:
    type: Artifact type of artifacts in this channel.
    optional: Whether providing this channel is optional.
    allow_empty: Optional. Only applicable to input channels. Whether having
      inputs to this channel is optional. Should only be explicitly set if
      `optional` is True. For backwards compatibility, if `optional` is True
      but this is not specified, we will treat this as True.
    is_async: Whether the channel parameter is for an intermediate output
      artifact or not. Defaults to False.
code: def __init__(
      self,
      type: Optional[
          Type[artifact.Artifact]
      ] = None,  # pylint: disable=redefined-builtin
      optional: bool = False,
      allow_empty: Optional[bool] = None,
      is_async: bool = False,
  ):
      """ChannelParameter constructor.
  
      Note the distinction between `optional` and `allow_empty`.
  
      An illustrative example: a component may make it optional to provide a
      specific channel C during pipeline definition time (i.e. set `optional` to
      True). However, it may require that if the channel C is provided during
      pipeline definition, during orchestration, the channel must have artifacts
      before the component is triggered (i.e. set `allow_empty` to False).
  
      Args:
        type: Artifact type of artifacts in this channel.
        optional: Whether providing this channel is optional.
        allow_empty: Optional. Only applicable to input channels. Whether having
          inputs to this channel is optional. Should only be explicitly set if
          `optional` is True. For backwards compatibility, if `optional` is True
          but this is not specified, we will treat this as True.
        is_async: Whether the channel parameter is for an intermediate output
          artifact or not. Defaults to False.
      """
      if not (
          inspect.isclass(type) and issubclass(type, artifact.Artifact)
      ):  # pytype: disable=wrong-arg-types
          raise ValueError(
              'Argument "type" of Channel constructor must be a subclass of '
              "tfx.types.Artifact."
          )
      self.type = type
      self.optional = optional
      self.is_async = is_async
  
      if allow_empty is None:
          # allow_empty not explicitly specified.
          # * If optional is True, then default to True for backwards compatibility.
          # * If optional is False, then set to False since the input should
          #   be mandatory.
          self.allow_empty_explicitly_set = False
          allow_empty = optional
      else:
          # allow_empty explicitly specified
          self.allow_empty_explicitly_set = True
          if not optional:
              raise ValueError(
                  "allow_empty should only be explicitly set if optional is True"
              )
      self.allow_empty = allow_empty
  
imports: import copy
  import inspect
  import itertools
  from typing import Any, Dict, List, Mapping, Optional, Type, cast
  from tfx.dsl.component.experimental.json_compat import check_strict_json_compat
  from tfx.dsl.placeholder import placeholder
  from tfx.types import artifact
  from tfx.types import channel
  from tfx.types.system_executions import SystemExecution
  from tfx.utils import json_utils
  from tfx.utils import proto_utils
  from google.protobuf import message
label_desc: Use image node_id 1 to create a new ChannelParameter object  with example: obj = ChannelParameter(type, optional, allow_empty, is_async)