def _validate()

in tfx/types/component_spec.py [0:0]
31 lines of code
12 McCabe index (conditional complexity)

  def _validate(cls):
    """Check the parameters and types passed to this ComponentSpec."""
    for param_name in ('PARAMETERS', 'INPUTS', 'OUTPUTS'):
      if not hasattr(cls, param_name):
        raise TypeError(f'{cls} does not have {param_name}.')
      param = getattr(cls, param_name)
      if not isinstance(param, Mapping):
        raise TypeError(
            f'{cls}.{param_name} should be a dict but got {param} instead.')

    # Validate that the ComponentSpec class is well-formed.
    seen_arg_names = set()
    for arg_name in itertools.chain(cls.PARAMETERS, cls.INPUTS, cls.OUTPUTS):
      if arg_name in seen_arg_names:
        raise ValueError(
            ('The ComponentSpec subclass %s has a duplicate argument with '
             'name %s. Argument names should be unique across the PARAMETERS, '
             'INPUTS and OUTPUTS dicts.') % (cls.__class__, arg_name))
      seen_arg_names.add(arg_name)

    for arg in cls.PARAMETERS.values():
      if not isinstance(arg, ExecutionParameter):
        raise TypeError(
            ('PARAMETERS dict expects values of type ExecutionParameter, '
             'got {}.').format(arg))

    for arg in itertools.chain(cls.INPUTS.values(), cls.OUTPUTS.values()):
      if not isinstance(arg, ChannelParameter):
        raise TypeError(
            ('INPUTS and OUTPUTS dicts expect values of type ChannelParameter, '
             ' got {}.').format(arg))

    for arg_name, arg in cls.OUTPUTS.items():
      if arg.allow_empty_explicitly_set:
        raise TypeError(
            'Output channels should not explicitly set allow_empty, but output '
            'channel %s did.' % arg_name)