class_docstr: Base class for function-based executors.
class_docstr_tok: ["Base","class","for","function-based","executors","."]
code: def Do(
      self,
      input_dict: Dict[str, List[tfx_types.Artifact]],
      output_dict: Dict[str, List[tfx_types.Artifact]],
      exec_properties: Dict[str, Any],
  ) -> None:
      function_args = _extract_func_args(
          obj=str(self),
          arg_formats=self._ARG_FORMATS,
          arg_defaults=self._ARG_DEFAULTS,
          input_dict=input_dict,
          output_dict=output_dict,
          exec_properties=exec_properties,
          beam_pipeline=self._make_beam_pipeline(),
      )
  
      # Call function and check returned values.
      outputs = self._FUNCTION(**function_args)
      outputs = outputs or {}
      output_dict.update(
          _assign_returned_values(
              function=self._FUNCTION,
              outputs=outputs,
              returned_values=self._RETURNED_VALUES,
              output_dict=output_dict,
              json_typehints=self._RETURN_JSON_COMPAT_TYPEHINT,
          )
      )
  
imports: import copy
  import functools
  import types
  import typing
  from typing import Any, Callable, ClassVar, Dict, List, Optional, Protocol, Type, Union
  from tfx import types
  from tfx.dsl.component.experimental import function_parser
  from tfx.dsl.component.experimental import json_compat
  from tfx.dsl.component.experimental import utils
  from tfx.dsl.components.base import base_beam_component
  from tfx.dsl.components.base import base_beam_executor
  from tfx.dsl.components.base import base_component
  from tfx.dsl.components.base import base_executor
  from tfx.dsl.components.base import executor_spec
  from tfx.types import channel
  from tfx.types import system_executions
label_desc: Use image node_id 1 for calling the _FunctionBeamExecutor obj's underlying member method code with example usage: obj.Do(input_dict, output_dict, exec_properties) without return types