Source code for json_api_doc.deserialization
import copy
[docs]def deserialize(content):
"""
:param content: A JSON API document already
:returns: The JSON API document parsed
"""
if "errors" in content:
return content
if "data" not in content:
raise AttributeError("This is not a JSON API document")
# be nondestructive with provided content
content = copy.deepcopy(content)
if "included" in content:
included = _parse_included(content["included"])
else:
included = {}
if isinstance(content["data"], dict):
return _resolve(_flat(content["data"]), included, set())
elif isinstance(content["data"], list):
result = []
for obj in content["data"]:
result.append(_resolve(_flat(obj), included, set()))
return result
else:
return None
def _resolve(data, included, resolved, deep=True):
if not isinstance(data, dict):
return data
keys = data.keys()
if keys == {"type", "id"} or keys == {"type", "id", "meta"}:
type_id = data["type"], data["id"]
meta = data.get("meta")
resolved_item = included.get(type_id, data)
resolved_item = resolved_item.copy()
if type_id not in resolved:
data = _resolve(
resolved_item,
included,
resolved | {type_id}
)
if meta is not None:
data = data.copy()
data.update(meta=meta)
return data
for key, value in data.items():
if isinstance(value, dict):
data[key] = _resolve(value, included, resolved)
elif isinstance(value, list):
if deep:
data[key] = [
_resolve(item, included, resolved, False)
for item in value
]
else:
data[key] = value
return data
def _parse_included(included):
result = {}
for include in included:
result[(include["type"], include["id"])] = _flat(include)
return result
def _flat(obj):
obj.pop("links", None)
obj.update(obj.pop("attributes", {}))
if "relationships" in obj:
for relationship, item in obj.pop("relationships").items():
data = item.get("data")
links = item.get("links")
if data is not None:
obj[relationship] = data
elif links:
obj[relationship] = item
else:
obj[relationship] = None
return obj