airflow/contrib/utils/gcp_field_validator.py
def _validate_field(self, validation_spec, dictionary_to_validate, parent=None,
                        force_optional=False):
        """
        Validates if field is OK.

        :param validation_spec: specification of the field
        :type validation_spec: dict
        :param dictionary_to_validate: dictionary where the field should be present
        :type dictionary_to_validate: dict
        :param parent: full path of parent field
        :type parent: str
        :param force_optional: forces the field to be optional
            (all union fields have force_optional set to True)
        :type force_optional: bool
        :return: True if the field is present
        """
        field_name = validation_spec['name']
        field_type = validation_spec.get('type')
        optional = validation_spec.get('optional')
        regexp = validation_spec.get('regexp')
        allow_empty = validation_spec.get('allow_empty')
        children_validation_specs = validation_spec.get('fields')
        required_api_version = validation_spec.get('api_version')
        custom_validation = validation_spec.get('custom_validation')

        full_field_path = self._get_field_name_with_parent(field_name=field_name,
                                                           parent=parent)
        if required_api_version and required_api_version != self._api_version:
            self.log.debug(
                "Skipping validation of the field '%s' for API version '%s' "
                "as it is only valid for API version '%s'",
                field_name, self._api_version, required_api_version)
            return False
        value = dictionary_to_validate.get(field_name)

        if (optional or force_optional) and value is None:
            self.log.debug("The optional field '%s' is missing. That's perfectly OK.", full_field_path)
            return False

        # Certainly down from here the field is present (value is not None)
        # so we should only return True from now on

        self._sanity_checks(children_validation_specs=children_validation_specs,
                            field_type=field_type,
                            full_field_path=full_field_path,
                            regexp=regexp,
                            allow_empty=allow_empty,
                            custom_validation=custom_validation,
                            value=value)

        if allow_empty is False:
            self._validate_is_empty(full_field_path, value)
        if regexp:
            self._validate_regexp(full_field_path, regexp, value)
        elif field_type == 'dict':
            if not isinstance(value, dict):
                raise GcpFieldValidationException(
                    "The field '{}' should be of dictionary type according to the "
                    "specification '{}' but it is '{}'".
                    format(full_field_path, validation_spec, value))
            if children_validation_specs is None:
                self.log.debug(
                    "The dict field '%s' has no nested fields defined in the "
                    "specification '%s'. That's perfectly ok - it's content will "
                    "not be validated.", full_field_path, validation_spec)
            else:
                self._validate_dict(children_validation_specs, full_field_path, value)
        elif field_type == 'union':
            if not children_validation_specs:
                raise GcpValidationSpecificationException(
                    "The union field '%s' has no nested fields "
                    "defined in specification '%s'. Unions should have at least one "
                    "nested field defined.", full_field_path, validation_spec)
            self._validate_union(children_validation_specs, full_field_path,
                                 dictionary_to_validate)
        elif field_type == 'list':
            if not isinstance(value, list):
                raise GcpFieldValidationException(
                    "The field '{}' should be of list type according to the "
                    "specification '{}' but it is '{}'".
                    format(full_field_path, validation_spec, value))
        elif custom_validation:
            try:
                custom_validation(value)
            except Exception as e:
                raise GcpFieldValidationException(
                    "Error while validating custom field '{}' specified by '{}': '{}'".
                    format(full_field_path, validation_spec, e))
        elif field_type is None:
            self.log.debug("The type of field '%s' is not specified in '%s'. "
                           "Not validating its content.", full_field_path, validation_spec)
        else:
            raise GcpValidationSpecificationException(
                "The field '{}' is of type '{}' in specification '{}'."
                "This type is unknown to validation!".format(
                    full_field_path, field_type, validation_spec))
        return True
def _validate_field(self, validation_spec, dictionary_to_validate, parent=None,
                        force_optional=False):
        """
        Validates if field is OK.

        :param validation_spec: specification of the field
        :type validation_spec: dict
        :param dictionary_to_validate: dictionary where the field should be present
        :type dictionary_to_validate: dict
        :param parent: full path of parent field
        :type parent: str
        :param force_optional: forces the field to be optional
            (all union fields have force_optional set to True)
        :type force_optional: bool
        :return: True if the field is present
        """
        field_name = validation_spec['name']
        field_type = validation_spec.get('type')
        optional = validation_spec.get('optional')
        regexp = validation_spec.get('regexp')
        allow_empty = validation_spec.get('allow_empty')
        children_validation_specs = validation_spec.get('fields')
        required_api_version = validation_spec.get('api_version')
        custom_validation = validation_spec.get('custom_validation')

        full_field_path = self._get_field_name_with_parent(field_name=field_name,
                                                           parent=parent)
        if required_api_version and required_api_version != self._api_version:
            self.log.debug(
                "Skipping validation of the field '%s' for API version '%s' "
                "as it is only valid for API version '%s'",
                field_name, self._api_version, required_api_version)
            return False
        value = dictionary_to_validate.get(field_name)

        if (optional or force_optional) and value is None:
            self.log.debug("The optional field '%s' is missing. That's perfectly OK.", full_field_path)
            return False

        # Certainly down from here the field is present (value is not None)
        # so we should only return True from now on

        self._sanity_checks(children_validation_specs=children_validation_specs,
                            field_type=field_type,
                            full_field_path=full_field_path,
                            regexp=regexp,
                            allow_empty=allow_empty,
                            custom_validation=custom_validation,
                            value=value)

        if allow_empty is False:
            self._validate_is_empty(full_field_path, value)
        if regexp:
            self._validate_regexp(full_field_path, regexp, value)
        elif field_type == 'dict':
            if not isinstance(value, dict):
                raise GcpFieldValidationException(
                    "The field '{}' should be of dictionary type according to the "
                    "specification '{}' but it is '{}'".
                    format(full_field_path, validation_spec, value))
            if children_validation_specs is None:
                self.log.debug(
                    "The dict field '%s' has no nested fields defined in the "
                    "specification '%s'. That's perfectly ok - it's content will "
                    "not be validated.", full_field_path, validation_spec)
            else:
                self._validate_dict(children_validation_specs, full_field_path, value)
        elif field_type == 'union':
            if not children_validation_specs:
                raise GcpValidationSpecificationException(
                    "The union field '%s' has no nested fields "
                    "defined in specification '%s'. Unions should have at least one "
                    "nested field defined.", full_field_path, validation_spec)
            self._validate_union(children_validation_specs, full_field_path,
                                 dictionary_to_validate)
        elif field_type == 'list':
            if not isinstance(value, list):
                raise GcpFieldValidationException(
                    "The field '{}' should be of list type according to the "
                    "specification '{}' but it is '{}'".
                    format(full_field_path, validation_spec, value))
        elif custom_validation:
            try:
                custom_validation(value)
            except Exception as e:
                raise GcpFieldValidationException(
                    "Error while validating custom field '{}' specified by '{}': '{}'".
                    format(full_field_path, validation_spec, e))
        elif field_type is None:
            self.log.debug("The type of field '%s' is not specified in '%s'. "
                           "Not validating its content.", full_field_path, validation_spec)
        else:
            raise GcpValidationSpecificationException(
                "The field '{}' is of type '{}' in specification '{}'."
                "This type is unknown to validation!".format(
                    full_field_path, field_type, validation_spec))
        return True
def _validate_field ( self , validation_spec , dictionary_to_validate , parent = None , force_optional = False ) : field_name = validation_spec [ 'name' ] field_type = validation_spec . get ( 'type' ) optional = validation_spec . get ( 'optional' ) regexp = validation_spec . get ( 'regexp' ) allow_empty = validation_spec . get ( 'allow_empty' ) children_validation_specs = validation_spec . get ( 'fields' ) required_api_version = validation_spec . get ( 'api_version' ) custom_validation = validation_spec . get ( 'custom_validation' ) full_field_path = self . _get_field_name_with_parent ( field_name = field_name , parent = parent ) if required_api_version and required_api_version != self . _api_version : self . log . debug ( "Skipping validation of the field '%s' for API version '%s' " "as it is only valid for API version '%s'" , field_name , self . _api_version , required_api_version ) return False value = dictionary_to_validate . get ( field_name ) if ( optional or force_optional ) and value is None : self . log . debug ( "The optional field '%s' is missing. That's perfectly OK." , full_field_path ) return False # Certainly down from here the field is present (value is not None) # so we should only return True from now on self . _sanity_checks ( children_validation_specs = children_validation_specs , field_type = field_type , full_field_path = full_field_path , regexp = regexp , allow_empty = allow_empty , custom_validation = custom_validation , value = value ) if allow_empty is False : self . _validate_is_empty ( full_field_path , value ) if regexp : self . _validate_regexp ( full_field_path , regexp , value ) elif field_type == 'dict' : if not isinstance ( value , dict ) : raise GcpFieldValidationException ( "The field '{}' should be of dictionary type according to the " "specification '{}' but it is '{}'" . format ( full_field_path , validation_spec , value ) ) if children_validation_specs is None : self . log . debug ( "The dict field '%s' has no nested fields defined in the " "specification '%s'. That's perfectly ok - it's content will " "not be validated." , full_field_path , validation_spec ) else : self . _validate_dict ( children_validation_specs , full_field_path , value ) elif field_type == 'union' : if not children_validation_specs : raise GcpValidationSpecificationException ( "The union field '%s' has no nested fields " "defined in specification '%s'. Unions should have at least one " "nested field defined." , full_field_path , validation_spec ) self . _validate_union ( children_validation_specs , full_field_path , dictionary_to_validate ) elif field_type == 'list' : if not isinstance ( value , list ) : raise GcpFieldValidationException ( "The field '{}' should be of list type according to the " "specification '{}' but it is '{}'" . format ( full_field_path , validation_spec , value ) ) elif custom_validation : try : custom_validation ( value ) except Exception as e : raise GcpFieldValidationException ( "Error while validating custom field '{}' specified by '{}': '{}'" . format ( full_field_path , validation_spec , e ) ) elif field_type is None : self . log . debug ( "The type of field '%s' is not specified in '%s'. " "Not validating its content." , full_field_path , validation_spec ) else : raise GcpValidationSpecificationException ( "The field '{}' is of type '{}' in specification '{}'." "This type is unknown to validation!" . format ( full_field_path , field_type , validation_spec ) ) return True
Validates if field is OK.

        :param validation_spec: specification of the field
        :type validation_spec: dict
        :param dictionary_to_validate: dictionary where the field should be present
        :type dictionary_to_validate: dict
        :param parent: full path of parent field
        :type parent: str
        :param force_optional: forces the field to be optional
            (all union fields have force_optional set to True)
        :type force_optional: bool
        :return: True if the field is present
Validates if field is OK .