Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def update_object(self,obj):
"""
Used to set values from a DataMessage that were updated (self.__updated == True),
as specified in the Attribute objects of the sub class of this, to the values of
the backend object that matches this DataMessage Object.
So in other words, this is the data mapping from DataMessage to backend object.
Read-Only Attributes are also skipped.
"""
janus_logger.debug("Starting to update object from DataMessage object.")
attributes = {attr:object.__getattribute__(self,attr)
for attr in dir(self)
if not callable(getattr(self,attr))
and type(object.__getattribute__(self,attr)) == Attribute
and issubclass(object.__getattribute__(self,attr).value_type,DataMessage) == False
and object.__getattribute__(self,attr).nested == False
and object.__getattribute__(self,attr).updated == True
and object.__getattribute__(self,attr).read_only == False
and object.__getattribute__(self,attr).mapping != None
#and object.__getattribute__(self,attr).name != 'id'
and not attr.startswith("__")}
for attr in attributes:
attr_obj = obj
attr_path = attributes[attr].mapping.split('.') #get mapping and split by '.', because this indicates a deeper path to get it.
if key_id != None:
type_name = relations[attr].value_type.__name__
if hasattr(relations[attr].value_type(),'type_name') and relations[attr].value_type().type_name != None: #if sub class has a member "type_name"...
type_name = relations[attr].value_type().type_name #get this type name
if isinstance(key_id,list): #one-to-many relation
relations[attr].key_value = {'data':[]}
for k in key_id:
relations[attr].key_value['data'].append({'type':type_name,'id':str(k)})
else: #one-to-one relation
relations[attr].key_value = {'data':{'type':type_name,'id':str(key_id)}}
if hasattr(self,'type_name') and self.type_name != None: #if sub class has a member "type_name"...
self.__type_name = self.type_name #... override __type_name to set this to 'type' in the final data object.
janus_logger.debug("Finished mapping object to message. ID: " + str(self.id) + " TYPE NAME: " + str(self.__type_name))
return self
if not callable(object.__getattribute__(self,attr))
and type(object.__getattribute__(self,attr)) == Attribute
and issubclass(object.__getattribute__(self,attr).value_type,DataMessage) == True
and object.__getattribute__(self,attr).nested == True
and not attr.startswith("__")
and object.__getattribute__(self,attr).name != 'id'
and object.__getattribute__(self,attr).key_value != None
}
if len(nested.keys()) > 0:
if 'relationships' in msg:
msg['relationships'].update(nested)
else:
msg['relationships'] = nested
janus_logger.debug("Transformed message object to dict.")
return msg
def map_object(self,obj,include_relationships=True,do_nesting=False):
"""
Used to set values from a python object, as specified in the Attribute objects
of the sub class of this, to the values of the Attribute objects of the sub class.
So in other words, this is the data mapping from object to DataMessage object.
"""
janus_logger.debug("Starting to map object to message.")
self.__data_object = obj #remember the object this message is based on
#get all members of the subclass containing Attribute members that are no relations as a dict.
#key => member name in the sub class.
#value => the Attribute inside of this member.
attributes = {attr:object.__getattribute__(self,attr)
for attr in dir(self)
if not callable(getattr(self,attr))
and type(object.__getattribute__(self,attr)) == Attribute
and issubclass(object.__getattribute__(self,attr).value_type,DataMessage) == False
and object.__getattribute__(self,attr).nested == False
and object.__getattribute__(self,attr).mapping != None
and object.__getattribute__(self,attr).write_only == False
and not attr.startswith("__")}
def map_message(self,message):
"""
Used to set values from a jsonapi request message, as specified in the Attribute objects
of the sub class of this, to the values of the Attribute objects of the sub class.
So in other words, this is the data mapping from message to DataMessage object.
"""
janus_logger.debug("Starting to map json message to DataMessage object.")
#get id
if 'id' in message:
self.id = message['id']
if 'attributes' in message:
#get attributes
attributes = {attr:object.__getattribute__(self,attr)
for attr in dir(self)
if not callable(getattr(self,attr))
and type(object.__getattribute__(self,attr)) == Attribute
and issubclass(object.__getattribute__(self,attr).value_type,DataMessage) == False
and object.__getattribute__(self,attr).nested == False
and object.__getattribute__(self,attr).mapping != None
and object.__getattribute__(self,attr).name != 'id'
and not attr.startswith("__")}
def wrapped_f(*a, **ka):
try:
#first check if this is not a HTTP OPTIONS call using a method defined based on the WS framework.
#if it is one return empty array and do nothing else.
if self.options_hook != None:
if self.options_hook() == True:
janus_logger.debug("This was an OPTIONS request.")
return {}
response_obj = f(*a, **ka)
#first check if there is an response object
#if not nothing to return so HTTP 204
#otherwise process response
if response_obj == None:
if self.before_send_hook != None:
self.before_send_hook(204,None,None)
janus_logger.debug("Decorated function returned None. Nothing to map.")
return None
else:
#check response object
if isinstance(response_obj,JanusResponse) == False:
janus_logger.info("Should map included: " + str(self.include_relationships))
if self.include_relationships:
included = self.__load_included(data,self.nest_in_responses)
#is there custome meta?
if response_obj.meta != None:
if self.meta == None:
self.meta = response_obj.meta
else:
self.meta.update(response_obj.meta)
message = JsonApiMessage(data=data,included=included,meta=self.meta,do_nesting=self.nest_in_responses).to_json() #render json response
#caching
if self.cached_set_hook != None and loaded_from_cache == False:
janus_logger.debug("Caching message")
self.cached_set_hook(response_obj,message)
if self.before_send_hook != None: #fire before send hook
self.before_send_hook(self.success_status,message,response_obj)
return message
except Exception as e:
err_msg = ErrorMessage.from_exception(e)
tb = traceback.format_exc()
if self.include_traceback_in_errors:
if err_msg.meta == None: err_msg.meta = {}
err_msg.traceback = tb
if self.error_hook != None:
self.error_hook(int(err_msg.status),err_msg,tb)