Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
metadata = document.pop("@metadata")
original_document = deepcopy(document)
original_metadata = deepcopy(metadata)
type_from_metadata = conventions.try_get_type_from_metadata(metadata)
mapper = conventions.mappers.get(object_type, None)
if object_type == dict:
return document, metadata, original_metadata, original_document
if type_from_metadata is None:
if object_type is not None:
metadata["Raven-Python-Type"] = "{0}.{1}".format(object_type.__module__, object_type.__name__)
else: # no type defined on document or during load, return a dict
return _DynamicStructure(**document), metadata, original_metadata, original_document
else:
object_from_metadata = Utils.import_class(type_from_metadata)
if object_from_metadata is not None:
if object_type is None:
object_type = object_from_metadata
elif Utils.is_inherit(object_type, object_from_metadata):
mapper = conventions.mappers.get(object_from_metadata, None) or mapper
object_type = object_from_metadata
if nested_object_types is None and mapper:
entity = create_entity_with_mapper(document, mapper, object_type)
else:
entity = _DynamicStructure(**document)
entity.__class__ = object_type
entity = Utils.initialize_object(document, object_type)
pass
elif isinstance(data[key], (tuple, list)):
for item in data[key]:
parse_dict_rec(item)
data[key] = mapper(key, data[key])
else:
mapper_result = mapper(key, data[key])
data[key] = mapper_result if mapper_result else data[key]
return data, True
except TypeError as e:
raise TypeError("Can't parse to custom object", e)
first_parsed, need_to_parse = parse_dict_rec(dict_obj)
# After create a complete dict for our object we need to create the object with the object_type
if first_parsed is not None and need_to_parse:
return Utils.initialize_object(first_parsed, object_type, convert_to_snake_case)
return first_parsed
def parse_dict_rec(data):
try:
if not isinstance(data, dict):
try:
for i in range(len(data)):
data[i] = Utils.initialize_object(parse_dict_rec(data[i])[0], object_type,
convert_to_snake_case)
except TypeError:
return data, False
for key in data:
if isinstance(data[key], dict):
parse_dict_rec(data[key])
data[key] = mapper(key, data[key])
if key is None:
pass
elif isinstance(data[key], (tuple, list)):
for item in data[key]:
parse_dict_rec(item)
data[key] = mapper(key, data[key])
else:
mapper_result = mapper(key, data[key])
def set_response(self, response):
if response is None:
return None
data = {}
try:
response = response.json()["Results"]
if len(response) > 1:
raise ValueError("response is Invalid")
for key, value in response[0].items():
data[Utils.convert_to_snake_case(key)] = value
return IndexDefinition(**data)
except ValueError:
raise response.raise_for_status()
def set_response(self, response):
if response is None:
return None
if response.status_code == 200:
attachment_details = {"content_type": response.headers.get("Content-Type", None),
"change_vector": Utils.get_change_vector_from_header(response),
"hash": response.headers.get("Attachment-Hash", None),
"size": response.headers.get("Attachment-Size", 0)}
return {"response": response, "details": attachment_details}
def _json_default(o):
if o is None:
return None
if isinstance(o, datetime):
return Utils.datetime_to_string(o)
elif isinstance(o, timedelta):
return Utils.timedelta_to_str(o)
elif getattr(o, "__dict__", None):
return o.__dict__
elif isinstance(o, set):
return list(o)
elif isinstance(o, int) or isinstance(o, float):
return str(o)
else:
raise TypeError(repr(o) + " is not JSON serializable (Try add a json default method to store convention)")