Testing

tfx.v1.testing

Public testing modules for TFX.

CLASS	DESCRIPTION
Channel	Dummy channel for testing.

Classes

Channel

Channel(artifact_type: Type[Artifact], artifact_ids: Sequence[int] = ())

Bases: BaseChannel

Dummy channel for testing.

METHOD	DESCRIPTION
as_optional	Creates an optional version of self.
future
get_data_dependent_node_ids	Get data dependent nodes of this channel.
no_trigger
trigger_by_property

ATTRIBUTE	DESCRIPTION
artifact_ids
input_trigger	TYPE: _InputTrigger
is_optional	If this is an "optional" channel. Changes Pipeline runtime behavior. TYPE: Optional[bool]
type	TYPE: Type[_AT]
type_name	Name of the artifact type class that Channel takes.

Source code in tfx/types/channel_utils.py

def __init__(
    self,
    artifact_type: Type[artifact.Artifact],
    artifact_ids: Sequence[int] = (),
):
  super().__init__(artifact_type)
  self.artifact_ids = artifact_ids

Attributes

artifact_ids instance-attribute

artifact_ids = artifact_ids

input_trigger property

input_trigger: _InputTrigger

is_optional property

is_optional: Optional[bool]

If this is an "optional" channel. Changes Pipeline runtime behavior.

type property writable

type: Type[_AT]

type_name property

type_name

Name of the artifact type class that Channel takes.

Functions

as_optional

as_optional() -> Self

Creates an optional version of self.

By default component input channels are considered required, meaning if the channel does not contain at least 1 artifact, the component will be skipped. Making channel optional disables this requirement and allows componenst to be executed with no artifacts from this channel.

RETURNS	DESCRIPTION
Self	A copy of self which is optional.

Source code in tfx/types/channel.py

def as_optional(self) -> typing_extensions.Self:
  """Creates an optional version of self.

  By default component input channels are considered required, meaning
  if the channel does not contain at least 1 artifact, the component
  will be skipped. Making channel optional disables this requirement and
  allows componenst to be executed with no artifacts from this channel.

  Returns:
    A copy of self which is optional.
  """
  new_channel = copy.copy(self)
  new_channel._is_optional = True  # pylint: disable=protected-access
  return new_channel

future

future() -> ChannelWrappedPlaceholder

Source code in tfx/types/channel_utils.py

def future(self) -> channel.ChannelWrappedPlaceholder:
  return channel.ChannelWrappedPlaceholder(self)

get_data_dependent_node_ids

get_data_dependent_node_ids() -> Set[str]

Get data dependent nodes of this channel.

Currently only the OutputChannel directly imposes the data dependency, but other channels can also indirectly have a data dependency if they depend on the OutputChannel. Use this abstract method to define transitive data dependency.

RETURNS	DESCRIPTION
Set[str]	A set of data-dependent node IDs.

Source code in tfx/types/channel_utils.py

def get_data_dependent_node_ids(self) -> Set[str]:
  return set()

no_trigger

no_trigger()

Source code in tfx/types/channel.py

@doc_controls.do_not_generate_docs
def no_trigger(self):
  return self._with_input_trigger(NoTrigger())

trigger_by_property

trigger_by_property(*property_keys: str)

Source code in tfx/types/channel.py

@doc_controls.do_not_generate_docs
def trigger_by_property(self, *property_keys: str):
  return self._with_input_trigger(TriggerByProperty(property_keys))