renderers.py

"""
Renderers
"""
importcopy
fromcollectionsimportdefaultdict
fromcollections.abcimportIterable

importinflection
fromdjango.db.modelsimportManager
fromdjango.templateimportloader
fromdjango.utils.encodingimportforce_str
fromrest_frameworkimportrelations, renderers
fromrest_framework.fieldsimportSkipField, get_attribute
fromrest_framework.relationsimportPKOnlyObject
fromrest_framework.serializersimportListSerializer, Serializer
fromrest_framework.settingsimportapi_settings

importrest_framework_json_api
fromrest_framework_json_apiimportutils
fromrest_framework_json_api.relationsimport (
HyperlinkedMixin,
ManySerializerMethodResourceRelatedField,
ResourceRelatedField,
SkipDataMixin,
)


classJSONRenderer(renderers.JSONRenderer):
"""
 The `JSONRenderer` exposes a number of methods that you may override if you need highly
 custom rendering control.

 Render a JSON response per the JSON:API spec:

 .. code-block:: json

 {
 "data": [
 {
 "type": "companies",
 "id": "1",
 "attributes": {
 "name": "Mozilla",
 "slug": "mozilla",
 "date-created": "2014-03-13 16:33:37"
 }
 }
 ]
 }
 """

media_type="application/vnd.api+json"
format="vnd.api+json"

@classmethod
defextract_attributes(cls, fields, resource):
"""
 Builds the `attributes` object of the JSON:API resource object.
 """
data= {}
forfield_name, fieldiniter(fields.items()):
# ID is always provided in the root of JSON:API so remove it from attributes
iffield_name=="id":
continue
# don't output a key for write only fields
iffields[field_name].write_only:
continue
# Skip fields with relations
ifutils.is_relationship_field(field):
continue

# Skip read_only attribute fields when `resource` is an empty
# serializer. Prevents the "Raw Data" form of the browsable API
# from rendering `"foo": null` for read only fields
try:
resource[field_name]
exceptKeyError:
iffields[field_name].read_only:
continue

data.update({field_name: resource.get(field_name)})

returnutils.format_field_names(data)

@classmethod
defextract_relationships(cls, fields, resource, resource_instance):
"""
 Builds the relationships top level object based on related serializers.
 """
# Avoid circular deps
fromrest_framework_json_api.relationsimportResourceRelatedField

data= {}

# Don't try to extract relationships from a non-existent resource
ifresource_instanceisNone:
return

forfield_name, fieldiniter(fields.items()):
# Skip URL field
iffield_name==api_settings.URL_FIELD_NAME:
continue

# don't output a key for write only fields
iffields[field_name].write_only:
continue

# Skip fields without relations
ifnotutils.is_relationship_field(field):
continue

source=field.source
relation_type=utils.get_related_resource_type(field)

ifisinstance(field, relations.HyperlinkedIdentityField):
resolved, relation_instance=utils.get_relation_instance(
resource_instance, source, field.parent
 )
ifnotresolved:
continue
# special case for HyperlinkedIdentityField
relation_data=list()

# Don't try to query an empty relation
relation_queryset= (
relation_instanceifrelation_instanceisnotNoneelselist()
 )

relation_data= [
 {"type": relation_type, "id": force_str(related_object.pk)}
forrelated_objectinrelation_queryset
 ]
data.update(
 {
field_name: {
"links": {"related": resource.get(field_name)},
"data": relation_data,
"meta": {"count": len(relation_data)},
 }
 }
 )
continue

relation_data= {}
ifisinstance(field, HyperlinkedMixin):
field_links=field.get_links(
resource_instance, field.related_link_lookup_field
 )
relation_data.update({"links": field_links} iffield_linkselsedict())
data.update({field_name: relation_data})

ifisinstance(field, (ResourceRelatedField,)):
ifnotisinstance(field, SkipDataMixin):
relation_data.update({"data": resource.get(field_name)})

ifisinstance(field, ManySerializerMethodResourceRelatedField):
relation_data.update(
 {"meta": {"count": len(resource.get(field_name))}}
 )

data.update({field_name: relation_data})
continue

ifisinstance(
field,
 (relations.PrimaryKeyRelatedField, relations.HyperlinkedRelatedField),
 ):
resolved, relation=utils.get_relation_instance(
resource_instance, f"{source}_id", field.parent
 )
ifnotresolved:
continue
relation_id=relationifresource.get(field_name) elseNone
relation_data= {"data": None}
ifrelation_idisnotNone:
relation_data["data"] = {
"type": relation_type,
"id": force_str(relation_id),
 }

ifisinstance(
field, relations.HyperlinkedRelatedField
 ) andresource.get(field_name):
relation_data.update(
 {"links": {"related": resource.get(field_name)}}
 )
data.update({field_name: relation_data})
continue

ifisinstance(field, relations.ManyRelatedField):
resolved, relation_instance=utils.get_relation_instance(
resource_instance, source, field.parent
 )
ifnotresolved:
continue

relation_data= {}

ifisinstance(resource.get(field_name), Iterable):
relation_data.update(
 {"meta": {"count": len(resource.get(field_name))}}
 )

ifisinstance(field.child_relation, ResourceRelatedField):
# special case for ResourceRelatedField
relation_data.update({"data": resource.get(field_name)})

ifisinstance(field.child_relation, HyperlinkedMixin):
field_links=field.child_relation.get_links(
resource_instance,
field.child_relation.related_link_lookup_field,
 )
relation_data.update(
 {"links": field_links} iffield_linkselsedict()
 )

data.update({field_name: relation_data})
continue

relation_data=list()
fornested_resource_instanceinrelation_instance:
nested_resource_instance_type= (
relation_type
orutils.get_resource_type_from_instance(
nested_resource_instance
 )
 )

relation_data.append(
 {
"type": nested_resource_instance_type,
"id": force_str(nested_resource_instance.pk),
 }
 )
data.update(
 {
field_name: {
"data": relation_data,
"meta": {"count": len(relation_data)},
 }
 }
 )
continue

returnutils.format_field_names(data)

@classmethod
defextract_relation_instance(cls, field, resource_instance):
"""
 Determines what instance represents given relation and extracts it.

 Relation instance is determined exactly same way as it determined
 in parent serializer
 """
try:
res=field.get_attribute(resource_instance)
ifisinstance(res, PKOnlyObject):
returnget_attribute(resource_instance, field.source_attrs)
returnres
exceptSkipField:
returnNone

@classmethod
defextract_included(
cls, fields, resource, resource_instance, included_resources, included_cache
 ):
"""
 Adds related data to the top level included key when the request includes
 ?include=example,example_field2
 """
# this function may be called with an empty record (example: Browsable Interface)
ifnotresource_instance:
return

current_serializer=fields.serializer
context=current_serializer.context
included_serializers=getattr(
current_serializer, "included_serializers", dict()
 )
included_resources=copy.copy(included_resources)
included_resources= [
inflection.underscore(value) forvalueinincluded_resources
 ]

forfield_name, fieldiniter(fields.items()):
# Skip URL field
iffield_name==api_settings.URL_FIELD_NAME:
continue

# Skip fields without relations
ifnotutils.is_relationship_field(field):
continue

try:
included_resources.remove(field_name)
exceptValueError:
# Skip fields not in requested included resources
# If no child field, directly continue with the next field
iffield_namenotin [
node.split(".")[0] fornodeinincluded_resources
 ]:
continue

relation_instance=cls.extract_relation_instance(field, resource_instance)
ifisinstance(relation_instance, Manager):
relation_instance=relation_instance.all()

serializer_data=resource.get(field_name)

ifisinstance(field, relations.ManyRelatedField):
serializer_class=included_serializers[field_name]
field=serializer_class(relation_instance, many=True, context=context)
serializer_data=field.data

ifisinstance(field, relations.RelatedField):
ifrelation_instanceisNoneornotserializer_data:
continue

many=field._kwargs.get("child_relation", None) isnotNone

ifisinstance(field, ResourceRelatedField) andnotmany:
already_included= (
serializer_data["type"] inincluded_cache
andserializer_data["id"]
inincluded_cache[serializer_data["type"]]
 )

ifalready_included:
continue

serializer_class=included_serializers[field_name]
field=serializer_class(relation_instance, many=many, context=context)
serializer_data=field.data

new_included_resources= [
key.replace(f"{field_name}.", "", 1)
forkeyinincluded_resources
iffield_name==key.split(".")[0]
 ]

ifisinstance(field, ListSerializer):
serializer=field.child
relation_type=utils.get_resource_type_from_serializer(serializer)
relation_queryset=list(relation_instance)

ifserializer_data:
forpositioninrange(len(serializer_data)):
serializer_resource=serializer_data[position]
nested_resource_instance=relation_queryset[position]
resource_type= (
relation_type
orutils.get_resource_type_from_instance(
nested_resource_instance
 )
 )
serializer_fields=utils.get_serializer_fields(
serializer.__class__(
nested_resource_instance, context=serializer.context
 )
 )
new_item=cls.build_json_resource_obj(
serializer_fields,
serializer_resource,
nested_resource_instance,
resource_type,
serializer,
getattr(serializer, "_poly_force_type_resolution", False),
 )
included_cache[new_item["type"]][new_item["id"]] =new_item

cls.extract_included(
serializer_fields,
serializer_resource,
nested_resource_instance,
new_included_resources,
included_cache,
 )

ifisinstance(field, Serializer):
relation_type=utils.get_resource_type_from_serializer(field)

# Get the serializer fields
serializer_fields=utils.get_serializer_fields(field)
ifserializer_data:
new_item=cls.build_json_resource_obj(
serializer_fields,
serializer_data,
relation_instance,
relation_type,
field,
getattr(field, "_poly_force_type_resolution", False),
 )
included_cache[new_item["type"]][new_item["id"]] =new_item

cls.extract_included(
serializer_fields,
serializer_data,
relation_instance,
new_included_resources,
included_cache,
 )

@classmethod
defextract_meta(cls, serializer, resource):
"""
 Gathers the data from serializer fields specified in meta_fields and adds it to
 the meta object.
 """
ifhasattr(serializer, "child"):
meta=getattr(serializer.child, "Meta", None)
else:
meta=getattr(serializer, "Meta", None)
meta_fields=getattr(meta, "meta_fields", [])
data= {}
forfield_nameinmeta_fields:
data.update({field_name: resource.get(field_name)})
returndata

@classmethod
defextract_root_meta(cls, serializer, resource):
"""
 Calls a `get_root_meta` function on a serializer, if it exists.
 """
many=False
ifhasattr(serializer, "child"):
many=True
serializer=serializer.child

data= {}
ifgetattr(serializer, "get_root_meta", None):
json_api_meta=serializer.get_root_meta(resource, many)
assertisinstance(json_api_meta, dict), "get_root_meta must return a dict"
data.update(json_api_meta)
returndata

@classmethod
defbuild_json_resource_obj(
cls,
fields,
resource,
resource_instance,
resource_name,
serializer,
force_type_resolution=False,
 ):
"""
 Builds the resource object (type, id, attributes) and extracts relationships.
 """
# Determine type from the instance if the underlying model is polymorphic
ifforce_type_resolution:
resource_name=utils.get_resource_type_from_instance(resource_instance)
resource_data= {
"type": resource_name,
"id": utils.get_resource_id(resource_instance, resource),
 }
attributes=cls.extract_attributes(fields, resource)
ifattributes:
resource_data["attributes"] =attributes
relationships=cls.extract_relationships(fields, resource, resource_instance)
ifrelationships:
resource_data["relationships"] =relationships
# Add 'self' link if field is present and valid
ifapi_settings.URL_FIELD_NAMEinresourceandisinstance(
fields[api_settings.URL_FIELD_NAME], relations.RelatedField
 ):
resource_data["links"] = {"self": resource[api_settings.URL_FIELD_NAME]}

meta=cls.extract_meta(serializer, resource)
ifmeta:
resource_data["meta"] =utils.format_field_names(meta)

returnresource_data

defrender_relationship_view(
self, data, accepted_media_type=None, renderer_context=None
 ):
# Special case for RelationshipView
view=renderer_context.get("view", None)
render_data= {"data": data}
links=view.get_links()
iflinks:
render_data["links"] =links
returnsuper().render(render_data, accepted_media_type, renderer_context)

defrender_errors(self, data, accepted_media_type=None, renderer_context=None):
returnsuper().render(
utils.format_errors(data), accepted_media_type, renderer_context
 )

defrender(self, data, accepted_media_type=None, renderer_context=None):
renderer_context=renderer_contextor {}

view=renderer_context.get("view", None)
request=renderer_context.get("request", None)

# Get the resource name.
resource_name=utils.get_resource_name(renderer_context)

# If this is an error response, skip the rest.
ifresource_name=="errors":
returnself.render_errors(data, accepted_media_type, renderer_context)

# if response.status_code is 204 then the data to be rendered must
# be None
response=renderer_context.get("response", None)
ifresponseisnotNoneandresponse.status_code==204:
returnsuper().render(None, accepted_media_type, renderer_context)

fromrest_framework_json_api.viewsimportRelationshipView

ifisinstance(view, RelationshipView):
returnself.render_relationship_view(
data, accepted_media_type, renderer_context
 )

# If `resource_name` is set to None then render default as the dev
# wants to build the output format manually.
ifresource_nameisNoneorresource_nameisFalse:
returnsuper().render(data, accepted_media_type, renderer_context)

json_api_data=data
# initialize json_api_meta with pagination meta or an empty dict
json_api_meta=data.get("meta", {}) ifisinstance(data, dict) else {}
included_cache=defaultdict(dict)

ifdataand"results"indata:
serializer_data=data["results"]
else:
serializer_data=data

serializer=getattr(serializer_data, "serializer", None)

included_resources=utils.get_included_resources(request, serializer)

ifserializerisnotNone:
# Extract root meta for any type of serializer
json_api_meta.update(self.extract_root_meta(serializer, serializer_data))

ifgetattr(serializer, "many", False):
json_api_data=list()

forpositioninrange(len(serializer_data)):
resource=serializer_data[position] # Get current resource
resource_instance=serializer.instance[
position
 ] # Get current instance

ifisinstance(
serializer.child,
rest_framework_json_api.serializers.PolymorphicModelSerializer,
 ):
resource_serializer_class= (
serializer.child.get_polymorphic_serializer_for_instance(
resource_instance
 )(context=serializer.child.context)
 )
else:
resource_serializer_class=serializer.child

fields=utils.get_serializer_fields(resource_serializer_class)
force_type_resolution=getattr(
resource_serializer_class, "_poly_force_type_resolution", False
 )

json_resource_obj=self.build_json_resource_obj(
fields,
resource,
resource_instance,
resource_name,
serializer,
force_type_resolution,
 )
json_api_data.append(json_resource_obj)

self.extract_included(
fields,
resource,
resource_instance,
included_resources,
included_cache,
 )
else:
fields=utils.get_serializer_fields(serializer)
force_type_resolution=getattr(
serializer, "_poly_force_type_resolution", False
 )

resource_instance=serializer.instance
json_api_data=self.build_json_resource_obj(
fields,
serializer_data,
resource_instance,
resource_name,
serializer,
force_type_resolution,
 )

self.extract_included(
fields,
serializer_data,
resource_instance,
included_resources,
included_cache,
 )

# Make sure we render data in a specific order
render_data= {}

ifisinstance(data, dict) anddata.get("links"):
render_data["links"] =data.get("links")

# format the api root link list
ifview.__class__andview.__class__.__name__=="APIRoot":
render_data["data"] =None
render_data["links"] =json_api_data
else:
render_data["data"] =json_api_data

ifincluded_cache:
ifisinstance(json_api_data, list):
objects=json_api_data
else:
objects= [json_api_data]

forobjectinobjects:
obj_type=object.get("type")
obj_id=object.get("id")
ifobj_typeinincluded_cacheandobj_idinincluded_cache[obj_type]:
delincluded_cache[obj_type][obj_id]
ifnotincluded_cache[obj_type]:
delincluded_cache[obj_type]

ifincluded_cache:
render_data["included"] =list()
forincluded_typeinsorted(included_cache.keys()):
forincluded_idinsorted(included_cache[included_type].keys()):
render_data["included"].append(
included_cache[included_type][included_id]
 )

ifjson_api_meta:
render_data["meta"] =utils.format_field_names(json_api_meta)

returnsuper().render(render_data, accepted_media_type, renderer_context)


classBrowsableAPIRenderer(renderers.BrowsableAPIRenderer):
template="rest_framework_json_api/api.html"
includes_template="rest_framework_json_api/includes.html"

defget_context(self, data, accepted_media_type, renderer_context):
context=super().get_context(data, accepted_media_type, renderer_context)
view=renderer_context["view"]

context["includes_form"] =self.get_includes_form(view)

returncontext

@classmethod
def_get_included_serializers(cls, serializer, prefix="", already_seen=None):
ifnotalready_seen:
already_seen=set()

ifserializerinalready_seen:
return []

included_serializers= []
already_seen.add(serializer)

forinclude, included_serializeringetattr(
serializer, "included_serializers", dict()
 ).items():
included_serializers.append(f"{prefix}{include}")
included_serializers.extend(
cls._get_included_serializers(
included_serializer,
f"{prefix}{include}.",
already_seen=already_seen,
 )
 )

returnincluded_serializers

defget_includes_form(self, view):
try:
if"related_field"inview.kwargs:
serializer_class=view.get_related_serializer_class()
else:
serializer_class=view.get_serializer_class()
exceptAttributeError:
return

ifnothasattr(serializer_class, "included_serializers"):
return

template=loader.get_template(self.includes_template)
context= {"elements": self._get_included_serializers(serializer_class)}
returntemplate.render(context)