code: def _extract_func_args(
      obj: str,
      arg_formats: Dict[str, int],
      arg_defaults: Dict[str, Any],
      input_dict: Dict[str, List[tfx_types.Artifact]],
      output_dict: Dict[str, List[tfx_types.Artifact]],
      exec_properties: Dict[str, Any],
      beam_pipeline: Optional[_BeamPipeline] = None,
  ) -> Dict[str, Any]:
      """Extracts function arguments for the decorated function."""
      result = {}
      for name, arg_format in arg_formats.items():
          if arg_format == utils.ArgFormats.INPUT_ARTIFACT:
              input_list = input_dict.get(name, [])
              if len(input_list) == 1:
                  result[name] = input_list[0]
              elif not input_list and name in arg_defaults:
                  # Do not pass the missing optional input.
                  pass
              else:
                  raise ValueError(
                      (
                          "Expected input %r to %s to be a singleton ValueArtifact channel "
                          "(got %s instead)."
                      )
                      % (name, obj, input_list)
                  )
          elif arg_format == utils.ArgFormats.LIST_INPUT_ARTIFACTS:
              result[name] = input_dict.get(name, [])
          elif arg_format == utils.ArgFormats.OUTPUT_ARTIFACT:
              output_list = output_dict.get(name, [])
              if len(output_list) == 1:
                  result[name] = output_list[0]
              else:
                  raise ValueError(
                      (
                          "Expected output %r to %s to be a singleton ValueArtifact channel "
                          "(got %s instead)."
                      )
                      % (name, obj, output_list)
                  )
          elif arg_format == utils.ArgFormats.ARTIFACT_VALUE:
              input_list = input_dict.get(name, [])
              if len(input_list) == 1:
                  result[name] = input_list[0].value
              elif not input_list and name in arg_defaults:
                  # Do not pass the missing optional input.
                  pass
              else:
                  raise ValueError(
                      (
                          "Expected input %r to %s to be a singleton ValueArtifact channel "
                          "(got %s instead)."
                      )
                      % (name, obj, input_list)
                  )
          elif arg_format == utils.ArgFormats.PARAMETER:
              if name in exec_properties:
                  result[name] = exec_properties[name]
              elif name in arg_defaults:
                  # Do not pass the missing optional input.
                  pass
              else:
                  raise ValueError(
                      (
                          "Expected non-optional parameter %r of %s to be provided, but no "
                          "value was passed."
                      )
                      % (name, obj)
                  )
          elif arg_format == utils.ArgFormats.BEAM_PARAMETER:
              result[name] = beam_pipeline
              if (
                  name in arg_defaults
                  and arg_defaults[name] is not None
              ):
                  raise ValueError(
                      "beam Pipeline parameter does not allow default ",
                      "value other than None.",
                  )
          else:
              raise ValueError(
                  "Unknown argument format: %r" % (arg_format,)
              )
      return result
  
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 a global function with example usage: _extract_func_args(obj, arg_formats, arg_defaults, input_dict, output_dict, exec_properties, beam_pipeline) and returns: result