def validate()

in src/sagemaker_algorithm_toolkit/channel_validation.py [0:0]

• 20 lines of code
• 9 McCabe index (conditional complexity)

    def validate(self, user_channels):
        """Validate the provided user-specified channels at runtime against the channels' supported configuration.

        Note that this adds default content type for channels if a default exists.

        :param user_channels: dictionary of channels formatted like so
                {
                    "channel_name": {
                        "ContentType": <content_type>.
                        "TrainingInputMode": <training_input_mode>,
                        "S3DistributionType": <s3_dist_type>,
                        ...
                    },
                    "channel_name": {...
                    }
                }
        """
        for channel in self.channels:
            if channel.name not in user_channels:
                if channel.required:
                    raise exc.UserError("Missing required channel: {}".format(channel.name))

        name_to_channel = {channel.name: channel for channel in self.channels}
        validated_channels = {}
        for channel, value in user_channels.items():
            try:
                channel_obj = name_to_channel[channel]
            except KeyError:
                raise exc.UserError("Extraneous channel found: {}".format(channel))

            if CONTENT_TYPE not in value:
                if self.default_content_type:
                    value[CONTENT_TYPE] = self.default_content_type
                else:
                    raise exc.UserError("Missing content type for channel: {}".format(channel))

            channel_obj.validate(value)
            validated_channels[channel] = value

        return validated_channels