Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Parameters
----------
c_dict : dict
the dictionary that contains the information
Returns
-------
callable
the reconstructed callable function or class
"""
c = None
if c_dict is not None:
if '_marshal' in c_dict:
if ObjectJSON.allow_marshal:
code = marshal.loads(base64.b64decode(c_dict['_marshal']))
c = types.FunctionType(code, globals(), code.co_name)
elif '_module' in c_dict:
module = c_dict['_module']
packages = module.split('.')
if packages[0] in ObjectJSON.safe_modules:
imp = importlib.import_module(module)
c = getattr(imp, c_dict['_name'])
return c
callable
the reconstructed callable function or class
"""
c = None
if c_dict is not None:
if '_marshal' in c_dict:
if ObjectJSON.allow_marshal:
code = marshal.loads(base64.b64decode(c_dict['_marshal']))
c = types.FunctionType(code, globals(), code.co_name)
elif '_module' in c_dict:
module = c_dict['_module']
packages = module.split('.')
if packages[0] in ObjectJSON.safe_modules:
imp = importlib.import_module(module)
c = getattr(imp, c_dict['_name'])
return c
',\n'.join(
map(lambda x: ' ' * 20 + x + '=' + x, global_vars)
) + ')' + '\n'
err += '\n and change your function definition like this'
err += '\n\n def ' + \
c.func_name + '(snapshot, ..., ' + \
'\n' + ',\n'.join(
map(lambda x: ' ' * 16 + x, global_vars)
) + '):'
unsafe_modules = [
module for module in import_vars
if module not in ObjectJSON.safe_modules
]
if ObjectJSON.prevent_unsafe_modules and len(unsafe_modules) > 0:
if len(err) > 0:
err += '\n\n'
err += 'The function you try to save requires the following' \
' modules to be installed: ' + str(unsafe_modules) + \
' which are not marked as safe! '
err += 'You can change the list of safe modules using '
err += '\n\n ObjectJSON.safe_modules.extend(['
err += '\n' + ',\n'.join(
map(lambda x: ' ' * 12 + x, unsafe_modules)
)
err += '\n ])'
err += '\n\n'
err += 'include the import statement in your function like'
err += '\n\n' + '\n'.join(
[' ' * 8 + 'import ' + v for v in unsafe_modules])
# is_class = isinstance(c, (type, types.ClassType))
# try saving known external classes of functions, e.g. `msmbuilder`
if root_module in ObjectJSON.safe_modules:
# only store the function/class and the module
return {
'_module': c.__module__,
'_name': c.__name__
}
# if the easy way did not work, try saving it using bytecode
if ObjectJSON.allow_marshal and callable(c):
# use marshal
global_vars = ObjectJSON._find_var(c, opcode.opmap['LOAD_GLOBAL'])
import_vars = ObjectJSON._find_var(c, opcode.opmap['IMPORT_NAME'])
all_builtins = dir(builtins)
global_vars = list(set(
[var for var in global_vars if var not in all_builtins]))
import_vars = list(set(import_vars))
err = ''
if len(global_vars) > 0:
err += 'The function you try to save relies on globally set ' \
'variables and these cannot be saved since storage ' \
'has no access to the global scope which includes ' \
'imports! \n\n'
err += 'We require that the following globals: ' + \
str(global_vars) + ' either\n'
return json_str
def from_json(self, json_string):
simplified = ujson.loads(json_string)
return self.build(simplified)
def unit_to_json(self, unit):
simple = self.unit_to_dict(unit)
return self.to_json(simple)
def unit_from_json(self, json_string):
return self.unit_from_dict(self.from_json(json_string))
class StorableObjectJSON(ObjectJSON):
def __init__(self, storage, unit_system=None):
super(StorableObjectJSON, self).__init__(unit_system)
self.excluded_keys = ['idx', 'json', 'identifier']
self.storage = storage
def simplify(self, obj, base_type=''):
if obj is self.storage:
return {'_storage': 'self'}
if obj.__class__.__module__ != builtin_module:
if obj.__class__ in self.storage._obj_store:
store = self.storage._obj_store[obj.__class__]
if not store.nestable or obj.base_cls_name != base_type:
# this also returns the base class name used for storage
# store objects only if they are not creatable. If so they
# will only be created in their top instance and we use
# the simplify from the super class ObjectJSON
if '_obj_uuid' in obj and '_store' in obj:
store = self.storage._stores[obj['_store']]
result = store.load(int(UUID(obj['_obj_uuid'])))
return result
if '_hex_uuid' in obj and '_store' in obj:
store = self.storage._stores[obj['_store']]
result = store.load(int(obj['_hex_uuid'].strip('L'), 16))
return result
return super(UUIDObjectJSON, self).build(obj)
class CachedUUIDObjectJSON(ObjectJSON):
def __init__(self, unit_system=None):
super(CachedUUIDObjectJSON, self).__init__(unit_system)
self.excluded_keys = ['json']
self.uuid_cache = WeakValueCache()
def simplify(self, obj, base_type=''):
if obj.__class__.__module__ != builtin_module:
if hasattr(obj, 'to_dict') and hasattr(obj, '__uuid__'):
# the object knows how to dismantle itself into a json string
if obj.__uuid__ not in self.uuid_cache:
self.uuid_cache[obj.__uuid__] = obj
return {
'_cls': obj.__class__.__name__,
'_obj_uuid': str(UUID(int=obj.__uuid__)),
'_dict': self.simplify(obj.to_dict(), base_type)}
err += '\n\n ObjectJSON.safe_modules.extend(['
err += '\n' + ',\n'.join(
map(lambda x: ' ' * 12 + x, unsafe_modules)
)
err += '\n ])'
err += '\n\n'
err += 'include the import statement in your function like'
err += '\n\n' + '\n'.join(
[' ' * 8 + 'import ' + v for v in unsafe_modules])
if len(err) > 0:
raise RuntimeError('Cannot store function! \n\n' +
word_wrap(err, 60))
return {
'_marshal': ObjectJSON._to_marshal(c),
'_global_vars': global_vars,
'_module_vars': import_vars
}
raise RuntimeError('Locally defined classes are not storable yet')