class_docstr: ResolverFunction represents a traceable function of resolver operators.
  
  Resolver function returns some form of channel depending on the function
  definition.
  
  It can return a single channel:
  
      trainer = Trainer(
          examples=latest_created(example_gen.outputs['examples']))
  
  or a dictionary of channels:
  
      k_fold_inputs = k_fold(example_gen.outputs['examples'], splits=5)
      trainer = Trainer(
          examples=k_fold_inputs['train'],
      )
      evaluator = Evaluator(
          examples=k_fold_inputs['eval'],
          model=trainer.outputs['model'],
      )
  
  or a ForEach-loopable channels:
  
    with ForEach(
        tfx.dsl.inputs.sequential_rolling_range(
            example_gens.outputs['examples'], n=3)) as train_window:
      trainer = Trainer(
          examples=train_window['examples'])
docstr: Temporarily patches output_type.
docstr_tok: ["Temporarily","patches","output_type","."]
code: def given_output_type(
      self, output_type: _TypeHint
  ) -> "ResolverFunction":
      """Temporarily patches output_type."""
      if not typing_utils.is_compatible(output_type, _TypeHint):
          raise ValueError(
              f"Invalid output_type: {output_type}, should be {_TypeHint}."
          )
      original = self._output_type
      try:
          self._output_type = output_type
          yield self
      finally:
          self._output_type = original
  
imports: import contextlib
  import inspect
  from typing import Callable, Type, Union, Mapping, Any, Optional, Sequence, cast, overload
  from tfx.dsl.control_flow import for_each_internal
  from tfx.dsl.input_resolution import resolver_op
  from tfx.types import artifact
  from tfx.types import channel
  from tfx.types import resolved_channel
  from tfx.utils import doc_controls
  from tfx.utils import typing_utils
label_desc: Use image node_id 2 for calling the ResolverFunction obj's underlying member method code with example usage: obj.given_output_type(output_type) without return types