src/sagemaker_algorithm_toolkit/channel_validation.py
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
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": {...
                    }
                }
        
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.
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 .
def validate(self, user_channels):
        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
def validate ( self , user_channels ) : 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
Validate the provided user-specified channels at runtime against the channels' supported configuration.
Validate the provided user - specified channels at runtime against the channels ' supported configuration .
"""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": {...
                    }
                }
        """
self
user_channels
dictionary of channels formatted like so
{
"channel_name": {
"ContentType": .
"TrainingInputMode": ,
"S3DistributionType": ,

},
"channel_name": {
}
}
dictionary of channels formatted like so { " channel_name " : { " ContentType " : . " TrainingInputMode " : " S3DistributionType " : } " channel_name " : { } }