def get_serializer(self, endpoint, collection):
""" Get `Serializer` class based on the given collection. """
return self.get_class(self.serializers, endpoint, collection)
python类Serializer()的实例源码
def generate_serializer(self, field_schema, name, model=None,
onmodel=False, model_serializers=None,
extra_serializers=None):
"""
Generate a `Serializer` class to serve the given `field_schema`.
There are two scenarios:
* If `field_schema` only consists of model fields and `onmodel` flag
is specified as True, then a `ModelSerializer` class is generated.
* If `field_schema` consists of non-model fields or a mixture of
model fields and non-model fields, then a `ContainerSerializer` class
is created.
"""
model_fields, extra_fields, sources = self._classify_fields(
field_schema)
if onmodel:
assert model, ('You cannot create a model serializer without'
' specifying its model')
serializer = generate_model_serializer(
name, model, model_fields, bases=model_serializers)
else:
serializer = generate_container_serializer(
model_fields, extra_fields, name, model,
instance_sources=sources,
model_serializers=model_serializers,
extra_serializers=extra_serializers)
return serializer
def construct_drf_collection(self, instance, spec, loc, context):
"""
Constructor for `.drf_collection` predicate.
It generates the required, `Serializer` class, and `ViewSet` class
based on the field schema, actions, permissions and additional
configuation (filter_fields, mixins) as specified on spec.
"""
endpoint = loc[0]
parent = context.get('parent_name')
constructed = context.get('constructed')
if '.collection' not in constructed:
raise doc.DeferConstructor
field_schema = doc.doc_get(instance, ('*',))
actions = doc.doc_get(instance, ('.actions', self.ADAPTER_CONF)) or []
model = self._get_or_import_model(loc[0] + '/' + parent,
loc + ('model',),
context.get('top_spec'))
model_serializers = spec.pop('model_serializers', [])
extra_serializers = spec.pop('serializers', [])
serializer = self.generate_serializer(
field_schema, parent, model=model,
model_serializers=model_serializers,
extra_serializers=extra_serializers)
kwargs = {k: v for k, v in spec.iteritems() if k != 'model'}
permissions = self.get_permissions(loc[:-1], context.get('top_spec'))
view = generate_view(parent, serializer, model, actions=actions,
permissions=permissions, **kwargs)
instance[self.ADAPTER_CONF] = view
self.serializers[endpoint + '/' + parent] = serializer
self.views[endpoint + '/' + parent] = view
return instance
def generate_nested_drf_field(self, instance, name, predicate_type, model,
onmodel=True, **kwargs):
"""
Generate a nested drf field, which is actually a `Serializer` class.
"""
kwargs.update(self.get_default_properties(predicate_type, kwargs))
field_schema = doc.doc_get(instance, (predicate_type,))
many = predicate_type == '.structarray'
model_serializers = kwargs.pop('model_serializers', [])
extra_serializers = kwargs.pop('serializers', [])
serializer = self.generate_serializer(
field_schema, name, onmodel=onmodel,
model_serializers=model_serializers,
extra_serializers=extra_serializers, model=model)
return serializer(many=many, **kwargs)
def __getitem__(self, key):
field = self.fields[key]
value = self.data.get(key)
error = self.errors.get(key) if hasattr(self, '_errors') else None
if isinstance(field, serializers.Serializer):
return serializers.NestedBoundField(field, value, error)
return serializers.BoundField(field, value, error)
def field_to_schema(field):
title = force_text(field.label) if field.label else ''
description = force_text(field.help_text) if field.help_text else ''
if isinstance(field, serializers.ListSerializer):
child_schema = field_to_schema(field.child)
return coreschema.Array(
items=child_schema,
title=title,
description=description
)
elif isinstance(field, serializers.Serializer):
return coreschema.Object(
properties=OrderedDict([
(key, field_to_schema(value))
for key, value
in field.fields.items()
]),
title=title,
description=description
)
elif isinstance(field, serializers.ManyRelatedField):
return coreschema.Array(
items=coreschema.String(),
title=title,
description=description
)
elif isinstance(field, serializers.RelatedField):
return coreschema.String(title=title, description=description)
elif isinstance(field, serializers.MultipleChoiceField):
return coreschema.Array(
items=coreschema.Enum(enum=list(field.choices.keys())),
title=title,
description=description
)
elif isinstance(field, serializers.ChoiceField):
return coreschema.Enum(
enum=list(field.choices.keys()),
title=title,
description=description
)
elif isinstance(field, serializers.BooleanField):
return coreschema.Boolean(title=title, description=description)
elif isinstance(field, (serializers.DecimalField, serializers.FloatField)):
return coreschema.Number(title=title, description=description)
elif isinstance(field, serializers.IntegerField):
return coreschema.Integer(title=title, description=description)
if field.style.get('base_template') == 'textarea.html':
return coreschema.String(
title=title,
description=description,
format='textarea'
)
return coreschema.String(title=title, description=description)
def run_autooptimization_discovery(serializer, prefix, select_related_set, prefetch_related_set, is_prefetch,
only_fields, include_fields):
if not hasattr(serializer, "Meta") or not hasattr(serializer.Meta, "model"):
return
model_class = serializer.Meta.model
if hasattr(serializer, "get_on_demand_fields"):
on_demand_fields = serializer.get_on_demand_fields()
else:
on_demand_fields = set()
def filter_field_name(field_name, fields_to_serialize):
if fields_to_serialize is not None:
return ContextPassing.filter_fields(field_name, fields_to_serialize)
return None
for field_name, field in serializer.fields.items():
if hasattr(serializer, "check_if_needs_serialization"):
if not serializer.check_if_needs_serialization(field_name, only_fields, include_fields, on_demand_fields):
continue
if isinstance(field, ListSerializer):
if "." not in field.source and hasattr(model_class, field.source):
model_field = getattr(model_class, field.source)
if check_if_prefetch_object(model_field):
prefetch_related_set.add(prefix + field.source)
run_autooptimization_discovery(field.child, prefix + field.source + "__", select_related_set,
prefetch_related_set, True,
filter_field_name(field_name, only_fields),
filter_field_name(field_name, include_fields))
elif isinstance(field, Serializer):
if "." not in field.source and hasattr(model_class, field.source):
model_field = getattr(model_class, field.source)
if check_if_related_object(model_field):
if is_prefetch:
prefetch_related_set.add(prefix + field.source)
else:
select_related_set.add(prefix + field.source)
run_autooptimization_discovery(field, prefix + field.source + "__", select_related_set,
prefetch_related_set, is_prefetch,
filter_field_name(field_name, only_fields),
filter_field_name(field_name, include_fields))
elif "." in field.source:
field_name = field.source.split(".", 1)[0]
if hasattr(model_class, field_name):
model_field = getattr(model_class, field_name)
if check_if_related_object(model_field):
select_related_set.add(prefix + field_name)
def get_serializer_fields(self, path, method, view, version=None, method_func=None):
"""
Return a list of `coreapi.Field` instances corresponding to any
request body input, as determined by the serializer class.
"""
if method not in ('PUT', 'PATCH', 'POST'):
return []
serializer_class = self.get_serializer_class(view, method_func)
if not serializer_class:
return []
serializer = serializer_class()
if isinstance(serializer, serializers.ListSerializer):
return [
Field(
name='data',
location='body',
required=True,
schema=coreschema.Array()
)
]
if not isinstance(serializer, serializers.Serializer):
return []
fields = []
for field in serializer.fields.values():
if field.read_only or isinstance(field, serializers.HiddenField):
continue
required = field.required and method != 'PATCH'
# if the attribute ('help_text') of this field is a lazy translation object, force it to generate a string
description = str(field.help_text) if isinstance(field.help_text, Promise) else field.help_text
fallback_schema = self.fallback_schema_from_field(field)
field = Field(
name=field.field_name,
location='form',
required=required,
schema=fallback_schema if fallback_schema else field_to_schema(field),
description=description,
)
fields.append(field)
return fields
def get_response_object(self, response_serializer_class, description):
fields = []
serializer = response_serializer_class()
nested_obj = {}
for field in serializer.fields.values():
# If field is a serializer, attempt to get its schema.
if isinstance(field, serializers.Serializer):
subfield_schema = self.get_response_object(field.__class__, None)[0].get('schema')
# If the schema exists, use it as the nested_obj
if subfield_schema is not None:
nested_obj[field.field_name] = subfield_schema
nested_obj[field.field_name]['description'] = field.help_text
continue
# Otherwise, carry-on and use the field's schema.
fallback_schema = self.fallback_schema_from_field(field)
fields.append(Field(
name=field.field_name,
location='form',
required=field.required,
schema=fallback_schema if fallback_schema else field_to_schema(field),
))
res = _get_parameters(Link(fields=fields), None)
if not res:
if nested_obj:
return {
'description': description,
'schema': {
'type': 'object',
'properties': nested_obj
}
}, {}
else:
return {}, {}
schema = res[0]['schema']
schema['properties'].update(nested_obj)
response_schema = {
'description': description,
'schema': schema
}
error_status_codes = {}
response_meta = getattr(response_serializer_class, 'Meta', None)
for status_code, description in getattr(response_meta, 'error_status_codes', {}).items():
error_status_codes[status_code] = {'description': description}
return response_schema, error_status_codes
def save(self, **kwargs):
ismodel = isinstance(self, ApimasModelSerializer)
assert not hasattr(self, 'save_object'), (
'Serializer `%s.%s` has old-style version 2 `.save_object()` '
'that is no longer compatible with REST framework 3. '
'Use the new-style `.create()` and `.update()` methods instead.' %
(self.__class__.__module__, self.__class__.__name__)
)
assert hasattr(self, '_errors'), (
'You must call `.is_valid()` before calling `.save()`.'
)
assert not self.errors, (
'You cannot call `.save()` on a serializer with invalid data.'
)
# Guard against incorrect use of `serializer.save(commit=False)`
assert 'commit' not in kwargs, (
"'commit' is not a valid keyword argument to the 'save()' method. "
"If you need to access data before committing to the database then "
"inspect 'serializer.validated_data' instead. "
"You can also pass additional keyword arguments to 'save()' if you "
"need to set extra attributes on the saved model instance. "
"For example: 'serializer.save(owner=request.user)'.'"
)
assert not hasattr(self, '_data'), (
"You cannot call `.save()` after accessing `serializer.data`."
"If you need to access data before committing to the database then "
"inspect 'serializer.validated_data' instead. "
)
if ismodel:
validated_data = dict(
list(self.validated_data.items()) +
list(kwargs.items())
)
else:
validated_data = self.validated_data
if self.instance is not None:
self.instance = self.update(self.instance, validated_data)\
if ismodel else self.update(self.instance, validated_data,
**kwargs)
if ismodel:
assert self.instance is not None, (
'`update()` did not return an object instance.'
)
else:
self.instance = self.create(validated_data) if ismodel\
else self.create(validated_data, **kwargs)
if ismodel:
assert self.instance is not None, (
'`create()` did not return an object instance.'
)
return self.instance
