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
python类missing()的实例源码
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'}
