def __init__()

in tfx/dsl/component/experimental/decorators.py [0:0]

• 35 lines of code
• 13 McCabe index (conditional complexity)

  def __init__(self, *unused_args, **kwargs):
    if unused_args:
      raise ValueError(('%s expects arguments to be passed as keyword '
                        'arguments') % (self.__class__.__name__,))
    spec_kwargs = {}
    unseen_args = set(kwargs.keys())
    for key, channel_parameter in self.SPEC_CLASS.INPUTS.items():
      if key not in kwargs and not channel_parameter.optional:
        raise ValueError('%s expects input %r to be a Channel of type %s.' %
                         (self.__class__.__name__, key, channel_parameter.type))
      if key in kwargs:
        spec_kwargs[key] = kwargs[key]
        unseen_args.remove(key)
    for key, parameter in self.SPEC_CLASS.PARAMETERS.items():
      if key not in kwargs and not parameter.optional:
        raise ValueError('%s expects parameter %r of type %s.' %
                         (self.__class__.__name__, key, parameter.type))
      if key in kwargs:
        spec_kwargs[key] = kwargs[key]
        unseen_args.remove(key)
    if unseen_args:
      raise ValueError(
          'Unknown arguments to %r: %s.' %
          (self.__class__.__name__, ', '.join(sorted(unseen_args))))
    for key, channel_parameter in self.SPEC_CLASS.OUTPUTS.items():
      artifact = channel_parameter.type()
      spec_kwargs[key] = channel.OutputChannel(artifact.type, self,
                                               key).set_artifacts([artifact])
      json_compat_typehint = getattr(channel_parameter, '_JSON_COMPAT_TYPEHINT',
                                     None)
      if json_compat_typehint:
        setattr(spec_kwargs[key], '_JSON_COMPAT_TYPEHINT', json_compat_typehint)
    spec = self.SPEC_CLASS(**spec_kwargs)
    super().__init__(spec)
    # Set class name, which is the decorated function name, as the default id.
    # It can be overwritten by the user.
    self._id = self.__class__.__name__