python类ValidationError()的实例源码

def create(name, description):
    """Create a new project.

    Example:

def _deserialize(self, value, attr, data):
        if not is_collection(value):
            value = [value]

        result = []
        errors = {}
        for idx, each in enumerate(value):
            try:
                result.append(self.container.deserialize(each))
            except ValidationError as e:
                if e.data is not None:
                    result.append(e.data)
                errors.update({idx: e.messages})

        if errors:
            raise ValidationError(errors, data=result)

        return result

def unwrap_item(self, item):
        """
        If the schema has an existing instance the id field must be set.
        :raises ValidationError: id field isn't present when required.
        """
        id_in_item = 'id' in item
        if self.load_existing and not id_in_item:
            # Updating Resources must include type and id keys
            # http://jsonapi.org/format/1.1/#crud-updating
            raise ma.ValidationError([{'detail': '`data` object must include `id` key.',
                                       'source': {'pointer': '/data'}}])

        if not self.load_existing and id_in_item:
            # Don't support client side identifier generation at this time.
            raise ForbiddenIdError()

        return super().unwrap_item(item)

def unwrap_request(self, data, many):
        if 'data' not in data:
            raise ma.ValidationError('Object must include `data` key.')

        data = data['data']
        data_is_collection = ma.utils.is_collection(data)
        if many:
            if not data_is_collection:
                raise ma.ValidationError([{'detail': '`data` object must be a collection.',
                                           'source': {'pointer': '/data'}}])

            return [self.unwrap_item(each) for each in data]

        if data_is_collection:
            raise ma.ValidationError([
                {'detail': '`data` object must be an object, not a collection.',
                 'source': {'pointer': '/data'}}])

        if data is None:
            # When updating relationships we need to specially handle the primary data being null.
            # http://jsonapi.org/format/1.1/#crud-updating-to-one-relationships
            raise NullPrimaryData()

        return self.unwrap_item(data)

def get_schema(self, *args, **kwargs):
        def dump(data, many=False, **kwargs):
            if many:
                return [{'id': i.id, 'name': i.name} for i in data], ''

            return {'id': data.id, 'name': data.name}, ''

        def load(data, partial=False):
            if not partial and data['id'] == 4:
                raise ma.ValidationError(message={'id': ['invalid value.']})
            return data, ''

        schema = mock.Mock()
        schema.dump = mock.Mock(side_effect=dump)
        schema.load = mock.Mock(side_effect=load)

        return schema

def _deserialize(self, value: str, attr, data: dict) -> Job.JobStatus:
        value_to_check = value.upper()
        if value_to_check == "REGISTERED":
            status = Job.JobStatus.REGISTERED
        elif value_to_check == "WORKING":
            status = Job.JobStatus.WORKING
        elif value_to_check == "COMPLETED":
            status = Job.JobStatus.COMPLETED
        elif value_to_check == "ERROR":
            status = Job.JobStatus.ERROR
        else:
            raise ValidationError(
                message='Unknown value for job status enum',
                fields=attr
            )
        return status

def validate_field_max_length(self, value, field_name, max_chars):
        """
        Validates maximum length for the field
        :param value: field value
        :type value: str
        :param field_name: name of the field
        :type field_name: str
        :param max_chars: maximum number of chars for the field
        :type max_chars: int
        :return:
        :raise ValidationError: if len(value) is greater than max_chars
        """
        if len(value) > max_chars:
            raise ValidationError(
                self.get_max_length_error_message(
                    field_name=field_name, max_chars=max_chars
                )
            )

def validate_field_is_one_of(value, field_name, choices):
        """
        Validates value is an element of choices
        :param value: field value
        :type value: str
        :param field_name: name of the field
        :type field_name: str
        :param choices: list of choices for the field
        :type choices: list of str
        :return:
        :raise ValidationError: if choices doesn't contain value
        """
        if value not in choices:
            raise ValidationError(
                'El campo "{}" es incorrecto'.format(field_name)
            )

def validate_all_fields(self, data):
        validation_errors = []
        for key in data.keys():
            underscore_key = convert_camel_case_to_underscore(key)
            validate_method = getattr(
                self, 'validate_{}'.format(underscore_key), None
            )
            try:
                if validate_method:
                    validate_method(data[key])
            except ValidationError as v:
                msg = get_error_message(
                    field_name=key, value=data[key], error_msg=v.message
                )
                validation_errors.append(msg)
        if validation_errors:
            raise ValidationError(validation_errors)

def validate_outpath(path):
    try:
        with tempfile.NamedTemporaryFile(mode='w', dir=path) as tfile:
            tfile.write('0')
            tfile.close()

    except Exception as e:
        if isinstance(e, OSError):
            if e.errno == errno.ENOENT:
                raise mm.ValidationError(
                    "%s is not in a directory that exists" % path)
            elif e.errno == errno.EACCES:
                raise mm.ValidationError(
                    "%s does not appear you can write to path" % path)
            else:
                raise mm.ValidationError(
                    "Unknown OSError: {}".format(e.message))
        else:
            raise mm.ValidationError(
                "Unknown Exception: {}".format(e.message))

def _validate(self, value):
        """

        Parameters
        ----------
        value : str
            filepath to validate you can write to that location

        Returns
        -------
        None

        Raises
        ------
        marshmallow.ValidationError
            If os.path.dirname cannot be applied, or if directory does not exist, or if you cannot write to that directory,
            or writing a temporary file there produces any crazy exception
        """
        try:
            path = os.path.dirname(value)
        except Exception as e:  # pragma: no cover
            raise mm.ValidationError(
                "%s cannot be os.path.dirname-ed" % value)  # pragma: no cover
        validate_outpath(path)

def _validate(self, value):
        if not os.path.isdir(value):
            try:
                os.makedirs(value)
                if self.mode is not None:
                    os.chmod(value, self.mode)
            except OSError as e:
                if e.errno == os.errno.EEXIST:
                    pass
                else:
                    raise mm.ValidationError(
                        "{} is not a directory and you cannot create it".format(
                            value)
                    )
        if self.mode is not None:
            try:
                assert((os.stat(value).st_mode & 0o777) == self.mode)
            except:
                raise mm.ValidationError(
                    "{} does not have the mode  ({}) that was specified ".format(
                        value, self.mode)
                )
        # use outputfile to test that a file in this location is a valid path
        validate_outpath(value)

def _validate(self, value):
        """

        Parameters
        ----------
        value : str
            value to validate

        """
        if (not hasattr(logging, value) or
                type(getattr(logging, value)) is not int):
            raise mm.ValidationError(
                    '{} is not a valid loglevel; try one of {}'.format(
                        value, LogLevel.options))

        # Would prefer this to be an argparse.Action subclass, but not yet sure how to implement this way
        logging.getLogger().setLevel(value)

def output(self,d,output_path=None):
        """method for outputing dictionary to the output_json file path after
        validating it through the output_schema_type

        Parameters
        ----------
        d:dict
            output dictionary to output 
        output_path: str
            path to save to output file, optional (with default to self.mod['output_json'] location)
        Raises
        ------
        marshmallow.ValidationError
            If any of the output dictionary doesn't meet the output schema
        """
        if output_path is None:
            output_path = self.args['output_json']

        output_json = self.get_output_json(d)
        with open(output_path,'w') as fp:
            json.dump(output_json,fp)

def test_bad_output(tmpdir):
    file_out = tmpdir.join('test_output_bad.json')
    input_parameters = {
        'output_json':str(file_out)
    }
    mod = ArgSchemaParser(input_data = input_parameters,
                          output_schema_type = MyOutputSchema,
                          args=[])
    M=[[5,5],[7,2]]
    Mnp = np.array(M)
    output = {
        "a":"example",
        "b":"not a number",
        "M":Mnp  
    }

    with pytest.raises(mm.ValidationError):
        mod.output(output)

def replace(cls, obj, field, value, state):
        """
        This is method for replace operation. It is separated to provide a
        possibility to easily override it in your Parameters.

        Args:
            obj (object): an instance to change.
            field (str): field name
            value (str): new value
            state (dict): inter-operations state storage

        Returns:
            processing_status (bool): True
        """
        if not hasattr(obj, field):
            raise ValidationError("Field '%s' does not exist, so it cannot be patched" % field)
        setattr(obj, field, value)
        return True

def set_pronouns(self, contact, original_data):
        subject_pronoun = original_data.get("pronoun")
        if subject_pronoun:
            if not isinstance(subject_pronoun, str):
                raise ValidationError("Pronoun must be a string")
            contact.pronouns = get_pronouns_by_subject(subject_pronoun)

        filters = original_data.get("pronouns")
        if filters:
            if isinstance(filters, list) and all(isinstance(f, str) for f in filters):
                contact.pronouns_list = [get_pronouns_by_subject(f) for f in filters]
            elif isinstance(filters, dict):
                contact.pronouns = get_pronouns_by_filters(filters)
            else:
                raise ValidationError(
                    "Pronouns must be a list of subject pronoun strings, "
                    "or an object of pronoun types."
                )

        return contact

def get_pronouns_by_subject(subject_pronoun):
    """
    Given a subject pronoun (as a string), return the Pronouns object
    that corresponds to that subject pronoun. This can be called with
    strings like "he", "she", and "it".

    If no Pronouns object is found that matches this subject pronoun, or
    there are multiple Pronouns objects that match, a ValidationError is raised.
    """
    try:
        return Pronouns.query.filter_by(subject=subject_pronoun).one()
    except NoResultFound:
        raise ValidationError(
            'No set of pronouns found for subject pronoun "{subject}"'.format(
                subject=subject_pronoun
            )
        )
    except MultipleResultsFound:
        raise ValidationError(
            'Multiple sets of pronouns found for subject pronoun '
            '"{subject}". Use more specific filters.'.format(
                subject=subject_pronoun
            )
        )

def get_pronouns_by_filters(filters):
    """
    Given a dictionary of pronoun filters, return the Pronouns object
    that corresponds to those pronouns pronoun. Some examples:

    {"subject": "he"}
    {"subject": "they", "reflexive": "themself"}
    {"subject": "they", "reflexive": "themselves"}
    {"possessive": "hers"}

    If no Pronouns object is found that matches these filters, or
    there are multiple Pronouns objects that match, a ValidationError is raised.
    """
    try:
        return Pronouns.query.filter_by(**filters).one()
    except NoResultFound:
        raise ValidationError(
            "No set of pronouns found for filters."
        )
    except MultipleResultsFound:
        raise ValidationError(
            "Multiple sets of pronouns found. Use more specific filters."
        )

def test_error_handling(
        self, exc, expected_error, expected_status_code, expected_message
    ):
        entrypoint = HttpEntrypoint('GET', 'url')
        entrypoint.expected_exceptions = (
            ValidationError,
            ProductNotFound,
            OrderNotFound,
            TypeError,
        )

        response = entrypoint.response_from_exception(exc)
        response_data = json.loads(response.data.decode())

        assert response.mimetype == 'application/json'
        assert response.status_code == expected_status_code
        assert response_data['error'] == expected_error
        assert response_data['message'] == expected_message

def add_product(self, request):
        """ Add product to cache in every region

        This endpoint can be called in any region and will dispatch event
        which will be handled by indexer's `handle_product_added`
        in all regions
        """
        try:
            payload = Product(strict=True).loads(
                request.get_data(as_text=True)
            ).data
        except ValidationError as err:
            return 400, json.dumps({
                'error': 'BAD_REQUEST',
                'message': err.messages
            })
        self.dispatch('product_added', Product(strict=True).dump(payload).data)
        return 200, ''

def order_product(self, request):
        """ HTTP entrypoint for ordering products

        This entrypoint can be called in any region but message will be
        published on a federated `fed.order_product` queue that is only
        consumed in `europe` region where master database and service with
        write permissions to it lives.
        """
        try:
            payload = Order(strict=True).loads(
                request.get_data(as_text=True)
            ).data
        except ValidationError as err:
            return 400, json.dumps({
                'error': 'BAD_REQUEST',
                'message': err.messages
            })

        self.order_product_publisher(
            payload, routing_key=ROUTING_KEY_ORDER_PRODUCT
        )
        return 200, ''

def _deserialize(self, value, attr, data):
        if value is None:
            return None
        if isinstance(value, Reference):
            if type(value) is not self.reference_cls:
                value = self.reference_cls(value.document_cls, value.pk)
            return value
        elif isinstance(value, self._document_implementation_cls):
            if not value.is_created:
                raise ValidationError(
                    _("Cannot reference a document that has not been created yet."))
            return self.reference_cls(value.__class__, value.pk)
        elif isinstance(value, dict):
            if value.keys() != {'cls', 'id'}:
                raise ValidationError(_("Generic reference must have `id` and `cls` fields."))
            try:
                _id = ObjectId(value['id'])
            except ValueError:
                raise ValidationError(_("Invalid `id` field."))
            return self._deserialize_from_mongo({
                '_cls': value['cls'],
                '_id': _id
            })
        else:
            raise ValidationError(_("Invalid value for generic reference field."))

def _deserialize(self, value, attr, data):
        embedded_document_cls = self.embedded_document_cls
        if isinstance(value, embedded_document_cls):
            return value
        # Handle inheritance deserialization here using `cls` field as hint
        if embedded_document_cls.opts.offspring and isinstance(value, dict) and 'cls' in value:
            to_use_cls_name = value.pop('cls')
            if not any(o for o in embedded_document_cls.opts.offspring
                       if o.__name__ == to_use_cls_name):
                raise ValidationError(_('Unknown document `{document}`.').format(
                    document=to_use_cls_name))
            try:
                to_use_cls = embedded_document_cls.opts.instance.retrieve_embedded_document(
                    to_use_cls_name)
            except NotRegisteredDocumentError as e:
                raise ValidationError(str(e))
            return to_use_cls(**value)
        else:
            # `Nested._deserialize` calls schema.load without partial=True
            data, errors = self.schema.load(value, partial=True)
            if errors:
                raise ValidationError(errors, data=data)
            return self._deserialize_from_mongo(data)

def load(self, data, partial=False):
        # Always use marshmallow partial load to skip required checks
        loaded_data, err = self.schema.load(data, partial=True)
        if err:
            raise ValidationError(err)
        self._data = loaded_data
        # Map the modified fields list on the the loaded data
        self.clear_modified()
        for key in loaded_data:
            self._mark_as_modified(key)
        if partial:
            self._collect_partial_fields(data)
        else:
            self.not_loaded_fields.clear()
        # Must be done last given it modify `loaded_data`
        self._add_missing_fields()

def test_dump_only(self):

        @self.instance.register
        class Doc(Document):
            dl = fields.IntField()
            do = fields.IntField(dump_only=True)
            lo = fields.IntField(load_only=True)
            nope = fields.IntField(dump_only=True, load_only=True)

        with pytest.raises(marshmallow.ValidationError):
            Doc(do=1)

        with pytest.raises(marshmallow.ValidationError):
            Doc(nope=1)

        assert Doc(dl=1, lo=2).dump() == {'dl': 1}

        assert Doc(nope=marshmallow.missing, do=marshmallow.missing).dump() == {}

def test_route_naming(self):

        class MySchema(EmbeddedSchema):
            in_front = fields.IntField(attribute='in_mongo')

        MyDataProxy = data_proxy_factory('My', MySchema())
        d = MyDataProxy()
        with pytest.raises(ValidationError):
            d.load({'in_mongo': 42})
        d.load({'in_front': 42})
        with pytest.raises(KeyError):
            d.get('in_mongo')
        assert d.get('in_front') == 42
        d.set('in_front', 24)
        assert d._data == {'in_mongo': 24}
        assert d.get('in_front') == 24
        assert d.dump() == {'in_front': 24}
        assert d.to_mongo() == {'in_mongo': 24}

def test_required(self):

        @self.instance.register
        class Person(Document):
            name = fields.StrField(required=True)
            birthday = fields.DateTimeField()

        person = Person(birthday=datetime(1968, 6, 9))

        # required should be called during commit
        with pytest.raises(ValidationError) as exc:
            person.required_validate()
        assert exc.value.messages == {'name': ['Missing data for required field.']}

        person.name = 'Marty'
        person.required_validate()

def test_strict_embedded_document(self):
        @self.instance.register
        class StrictEmbeddedDoc(EmbeddedDocument):
            a = fields.IntField()

        @self.instance.register
        class NonStrictEmbeddedDoc(EmbeddedDocument):
            a = fields.IntField()

            class Meta:
                strict = False

        data_with_bonus = {'a': 42, 'b': 'foo'}
        with pytest.raises(exceptions.UnknownFieldInDBError):
            StrictEmbeddedDoc.build_from_mongo(data_with_bonus)

        non_strict_doc = NonStrictEmbeddedDoc.build_from_mongo(data_with_bonus)
        assert non_strict_doc.to_mongo() == data_with_bonus
        non_strict_doc.dump() == {'a': 42}

        with pytest.raises(exceptions.ValidationError) as exc:
            NonStrictEmbeddedDoc(a=42, b='foo')
        assert exc.value.messages == {'_schema': ['Unknown field name b.']}

def test_phone_number_field_load(number, expected, strict, passes):
    """Ensure that the Phone Number field can format numbers correctly."""
    schema = PhoneNumberSchema(context={'strict_phone_validation': strict})
    payload = {'phone': number}

    if passes:
        deserialized = schema.load(payload).data
        assert deserialized['phone'] == expected

        serialized = schema.dump(deserialized).data
        assert serialized['phone'] == expected
    else:
        error_msg = {
            'phone': [("The value for phone ({}) is not a valid phone "
                       "number.".format(number))]
        }

        with pytest.raises(ValidationError, message=error_msg):
            schema.load(payload)