Python marshmallow 模块，missing() 实例源码

def parse(self, data):
        """Parse operator and value from filter's data."""
        val = data.get(self.fname, missing)
        if not isinstance(val, dict):
            val = self.field.deserialize(val)
            request.filters[self.fname] = val
            return (self.operators['$eq'], val),

        ops = ()
        request.filters[self.fname] = {}
        for op, val in val.items():
            if op not in self.operators:
                continue
            val = self.field.deserialize(val) if op not in self.list_ops else [self.field.deserialize(v) for v in val]  # noqa
            ops += (self.operators[op], val),
            request.filters[self.fname][op] = val

        return ops

def make_object(self, in_data):
        """marshmallow.pre_load decorated function for applying defaults on deserialation

        Parameters
        ----------
        in_data :


        Returns
        -------
        dict
            a dictionary with default values applied

        """
        for name, field in self.fields.items():
            if name not in in_data:
                if field.default is not mm.missing:
                    in_data[name] = field.default
        return in_data

def _to_mongo_update(self):
        mongo_data = {}
        set_data = {}
        unset_data = []
        for name, field in self._fields.items():
            name = field.attribute or name
            v = self._data[name]
            if name in self._modified_data or (
                    isinstance(v, BaseDataObject) and v.is_modified()):
                v = field.serialize_to_mongo(v)
                if v is missing:
                    unset_data.append(name)
                else:
                    set_data[name] = v
        if set_data:
            mongo_data['$set'] = set_data
        if unset_data:
            mongo_data['$unset'] = {k: "" for k in unset_data}
        return mongo_data or None

def schema_from_umongo_get_attribute(self, attr, obj, default):
    """
    Overwrite default `Schema.get_attribute` method by this one to access
        umongo missing fields instead of returning `None`.

    example::

        class MySchema(marshsmallow.Schema):
            get_attribute = schema_from_umongo_get_attribute

            # Define the rest of your schema
            ...

    """
    ret = MaSchema.get_attribute(self, attr, obj, default)
    if ret is None and ret is not default and attr in obj.schema.fields:
        raw_ret = obj._data.get(attr)
        return default if raw_ret is missing else raw_ret
    else:
        return ret

def test_keep_attributes(self):
        @self.instance.register
        class Vehicle(Document):
            brand = fields.StrField(description='Manufacturer name')
            category = fields.StrField(required=True)
            nb_wheels = fields.IntField(missing=4)

        ma_schema_cls = Vehicle.schema.as_marshmallow_schema()
        schema = ma_schema_cls()

        ret = schema.load({})
        assert ret.errors == {'category': ['Missing data for required field.']}
        ret = schema.load({'category': 'Car'})
        assert ret.data == {'category': 'Car', 'nb_wheels': 4}

        assert schema.fields['brand'].metadata['description'] == 'Manufacturer name'

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_simple(self):

        class MySchema(EmbeddedSchema):
            a = fields.IntField()
            b = fields.IntField()

        MyDataProxy = data_proxy_factory('My', MySchema())
        d = MyDataProxy()
        d.load({'a': 1, 'b': 2})
        assert d.get('a') == 1
        d.set('b', 3)
        assert d.get('b') == 3
        assert d._data == {'a': 1, 'b': 3}
        assert d.dump() == {'a': 1, 'b': 3}
        d.delete('b')
        assert d._data == {'a': 1, 'b': missing}
        assert d.dump() == {'a': 1}

def test_equality(self):

        class MySchema(EmbeddedSchema):
            a = fields.IntField()
            b = fields.IntField(attribute='in_mongo_b')

        MyDataProxy = data_proxy_factory('My', MySchema())
        d1 = MyDataProxy()
        d1.load({'a': 1, 'b': 2})
        assert d1 == {'a': 1, 'in_mongo_b': 2}

        d2 = MyDataProxy()
        d2.load({'a': 1, 'b': 2})
        assert d1 == d2

        assert d1 != None  # noqa: E711 (None comparison)
        assert d1 != missing
        assert None != d1
        assert missing != d1

def test_set_to_missing_fields(self):

        class MySchema(EmbeddedSchema):
            a = fields.IntField()
            b = fields.IntField(attribute='in_mongo_b')

        MyDataProxy = data_proxy_factory('My', MySchema())
        d = MyDataProxy(data={'a': 1})
        assert d.get('b') is missing
        assert d.get_by_mongo_name('in_mongo_b') is missing
        assert d._data['in_mongo_b'] is missing
        d.set('b', 2)
        assert d.get('b') == 2
        d.delete('b')
        # Can do it two time in a row without error
        d.delete('b')
        assert d._data['in_mongo_b'] is missing

def test_equality(self):
        @self.instance.register
        class MyChildEmbeddedDocument(EmbeddedDocument):
            num = fields.IntField()

        @self.instance.register
        class MyParentEmbeddedDocument(EmbeddedDocument):
            embedded = fields.EmbeddedField(MyChildEmbeddedDocument)

        emb_1 = MyParentEmbeddedDocument(embedded={'num': 1})
        emb_2 = MyParentEmbeddedDocument(embedded={'num': 1})
        emb_3 = MyParentEmbeddedDocument(embedded={})
        emb_4 = MyParentEmbeddedDocument()

        assert emb_1 == emb_2
        assert emb_1 != emb_3
        assert emb_1 != emb_4
        assert emb_1 != None  # noqa: E711 (None comparison)
        assert emb_1 != missing
        assert None != emb_1
        assert missing != emb_1

def _field2method(field_or_factory, method_name, preprocess=None):
    def method(self, name, default=ma.missing, subcast=None, **kwargs):
        name = self._prefix + name if self._prefix else name
        missing = kwargs.pop('missing', None) or default
        if isinstance(field_or_factory, type) and issubclass(field_or_factory, ma.fields.Field):
            field = field_or_factory(missing=missing, **kwargs)
        else:
            field = field_or_factory(subcast=subcast, missing=missing, **kwargs)
        self._fields[name] = field
        parsed_key, raw_value = _get_from_environ(name, ma.missing)
        if raw_value is ma.missing and field.missing is ma.missing:
            raise EnvError('Environment variable "{}" not set'.format(parsed_key))
        value = raw_value or field.missing
        if preprocess:
            value = preprocess(value, subcast=subcast, **kwargs)
        try:
            value = field.deserialize(value)
        except ma.ValidationError as err:
            raise EnvError('Environment variable "{}" invalid: {}'.format(name, err.args[0]))
        else:
            self._values[name] = value
            return value
    method.__name__ = str(method_name)  # cast to str for Py2 compat
    return method

def _serialize(self, value, attr, obj):
        if value is None:
            return ma.missing
        return str(value)

def _from_python_type(field, pytype):
    json_schema = {
        'title': field.attribute or field.name,
    }
    for key, val in TYPE_MAP[pytype].items():
        json_schema[key] = val
    if field.default is not missing:
        json_schema['default'] = field.default

    return json_schema

def serialize(self, attr_name, field_symbol,
                  assignment_template, field_obj):
        # type: (str, str, str, fields.Field) -> IndentedString
        body = IndentedString()
        if self.context.is_serializing:
            default_str = 'default'
            default_value = field_obj.default
        else:
            default_str = 'missing'
            default_value = field_obj.missing
            if field_obj.required:
                body += assignment_template.format('obj["{attr_name}"]'.format(
                    attr_name=attr_name))
                return body
        if default_value == missing:
            body += 'if "{attr_name}" in obj:'.format(attr_name=attr_name)
            with body.indent():
                body += assignment_template.format('obj["{attr_name}"]'.format(
                    attr_name=attr_name))
        else:
            if callable(default_value):
                default_str += '()'

            body += assignment_template.format(
                'obj.get("{attr_name}", {field_symbol}__{default_str})'.format(
                    attr_name=attr_name, field_symbol=field_symbol,
                    default_str=default_str))
        return body

def generate_unmarshall_method(schema, context=missing):
    context = context or JitContext()
    context.is_serializing = False
    return generate_marshall_method(schema, context)

def fill_defaults(schema, args):
    """DEPRECATED, function to fill in default values from schema into args
    bug: goes into an infinite loop when there is a recursively defined schema

    Parameters
    ----------
    schema : marshmallow.Schema
        schema to get defaults from
    args :


    Returns
    -------
    dict
        dictionary with missing default values filled in

    """

    defaults = []

    # find all of the schema entries with default values
    schemata = [(schema, [])]
    while schemata:
        subschema, path = schemata.pop()
        for k, v in subschema.declared_fields.items():
            if isinstance(v, mm.fields.Nested):
                schemata.append((v.schema, path + [k]))
            elif v.default != mm.missing:
                defaults.append((path + [k], v.default))

    # put the default entries into the args dictionary
    args = copy.deepcopy(args)
    for path, val in defaults:
        d = args
        for path_item in path[:-1]:
            d = d.setdefault(path_item, {})
        if path[-1] not in d:
            d[path[-1]] = val
    return args

def _serialize(self, value, key, obj):
        """Output the URL for the endpoint, given the kwargs passed to
        ``__init__``.
        """
        param_values = {}
        for name, attr_tpl in iteritems(self.params):
            attr_name = _tpl(str(attr_tpl))
            if attr_name:
                attribute_value = utils.get_value(attr_name, obj, default=missing)
                if attribute_value is not missing:
                    param_values[name] = attribute_value
                else:
                    err = AttributeError(
                        '{attr_name!r} is not a valid '
                        'attribute of {obj!r}'.format(attr_name=attr_name, obj=obj)
                    )
                    if has_forced_error:
                        raise ForcedError(err)
                    else:
                        raise err
            else:
                param_values[name] = attr_tpl
        try:
            return url_for(self.endpoint, **param_values)
        except BuildError as err:  # Make sure BuildErrors are raised
            if has_forced_error:
                raise ForcedError(err)
            else:
                raise err

def __init__(self, *args, **kwargs):
        kwargs.setdefault('default', {})
        kwargs.setdefault('missing', Dict)
        super().__init__(*args, **kwargs)

def _serialize_to_mongo(self, obj):
        if not obj:
            return missing
        return dict(obj)

def __init__(self, *args, **kwargs):
        kwargs.setdefault('default', [])
        kwargs.setdefault('missing', lambda: List(self.container))
        super().__init__(*args, **kwargs)

def _serialize_to_mongo(self, obj):
        if not obj:
            return missing
        return [self.container.serialize_to_mongo(each) for each in obj]

def _required_validate(self, value):
        if value is missing or not hasattr(self.container, '_required_validate'):
            return
        required_validate = self.container._required_validate
        errors = {}
        for i, sub_value in enumerate(value):
            try:
                required_validate(sub_value)
            except ValidationError as exc:
                errors[i] = exc.messages
        if errors:
            raise ValidationError(errors)

def _required_validate(self, value):
        if value is not missing:
            value.required_validate()

def _to_mongo(self):
        mongo_data = {}
        for k, v in self._data.items():
            field = self._fields_from_mongo_key[k]
            v = field.serialize_to_mongo(v)
            if v is not missing:
                mongo_data[k] = v
        return mongo_data

def delete_by_mongo_name(self, name):
        self.set_by_mongo_name(name, missing)

def get(self, name, to_raise=KeyError):
        name, field = self._get_field(name, to_raise)
        value = self._data[name]
        if value is missing and field.default is not missing:
            return field.default
        return value

def _add_missing_fields(self):
        # TODO: we should be able to do that by configuring marshmallow...
        for name, field in self._fields.items():
            mongo_name = field.attribute or name
            if mongo_name not in self._data:
                if callable(field.missing):
                    self._data[mongo_name] = field.missing()
                else:
                    self._data[mongo_name] = field.missing

def required_validate(self):
        errors = {}
        for name, field in self.schema.fields.items():
            value = self._data[field.attribute or name]
            if field.required and value is missing:
                errors[name] = [_("Missing data for required field.")]
            elif hasattr(field, '_required_validate'):
                try:
                    field._required_validate(value)
                except ValidationError as exc:
                    errors[name] = exc.messages
        if errors:
            raise ValidationError(errors)

    # Standards iterators providing oo and mongo worlds views

def schema_validator_check_unknown_fields(self, data, original_data):
    """
    Schema validator, raise ValidationError for unknown fields in a
    marshmallow schema.

    example::

        class MySchema(marshsmallow.Schema):
            # method's name is not important
            __check_unknown_fields = validates_schema(pass_original=True)(
                schema_validator_check_unknown_fields)

            # Define the rest of your schema
            ...

    ..note:: Unknown fields with `missing` value will be ignored
    """
    # Just skip if dummy data have been passed to the schema
    if not isinstance(original_data, dict):
        return
    loadable_fields = [k for k, v in self.fields.items() if not v.dump_only]
    unknown_fields = {key for key, value in original_data.items()
                      if value is not missing and key not in loadable_fields}
    if unknown_fields:
        raise ValidationError([_('Unknown field name {field}.').format(field=field)
                               for field in unknown_fields])

def test_default(self):

        class MySchema(EmbeddedSchema):
            with_default = fields.StrField(default='default_value')
            with_missing = fields.StrField(missing='missing_value')

        MyDataProxy = data_proxy_factory('My', MySchema())
        d = MyDataProxy(data={})
        assert d._data['with_default'] is missing
        assert d._data['with_missing'] is 'missing_value'
        assert d.get('with_default') == 'default_value'
        assert d.get('with_missing') == 'missing_value'
        assert d.to_mongo() == {'with_missing': 'missing_value'}
        assert d.dump() == {'with_default': 'default_value', 'with_missing': 'missing_value'}

def _func2method(func, method_name):
    def method(self, name, default=ma.missing, subcast=None, **kwargs):
        name = self._prefix + name if self._prefix else name
        parsed_key, raw_value = _get_from_environ(name, default)
        if raw_value is ma.missing:
            raise EnvError('Environment variable "{}" not set'.format(parsed_key))
        value = func(raw_value, **kwargs)
        self._fields[name] = ma.fields.Field(**kwargs)
        self._values[name] = value
        return value
    method.__name__ = str(method_name)  # cast to str for Py2 compat
    return method