def marshal_load(f):
if isinstance(f, file):
return marshal.load(f)
return marshal.loads(f.read())
python类loads()的实例源码
def marshal_load(f):
if isinstance(f, file):
return marshal.load(f)
return marshal.loads(f.read())
def loads(str):
file = StringIO(str)
return Unpickler(file).load()
# Doctest
def unserialize_key(self, serialized_key):
if not serialized_key:
raise RuntimeError("Invalid serialized key")
return marshal.loads(serialized_key)
def load_exists_positions(self):
"""Load all keys from memory
"""
self._syncs += 1
self._used = self.get_area_size()
self._sign = self.get_area_sign()
for _, (key, _), positions in self.read_memory():
self._positions[key] = positions
#self._keys_cache[marshal.loads(key)] = key
def marshal_load(f):
if isinstance(f, file):
return marshal.load(f)
return marshal.loads(f.read())
def marshal_load(f):
if isinstance(f, file):
return marshal.load(f)
return marshal.loads(f.read())
def marshal_load(f):
if isinstance(f, file):
return marshal.load(f)
return marshal.loads(f.read())
def pupy_add_package(pkdic):
""" update the modules dictionary to allow remote imports of new packages """
import cPickle
global modules
modules.update(cPickle.loads(pkdic))
def _read_with_length(self, stream):
length = read_int(stream)
if length == SpecialLengths.END_OF_DATA_SECTION:
raise EOFError
elif length == SpecialLengths.NULL:
return None
obj = stream.read(length)
if len(obj) < length:
raise EOFError
return self.loads(obj)
def loads(self, obj):
"""
Deserialize an object from a byte array.
"""
raise NotImplementedError
def loads(self, obj):
return obj
def loads(self, obj, encoding="bytes"):
return pickle.loads(obj, encoding=encoding)
def loads(self, obj, encoding=None):
return pickle.loads(obj)
def loads(self, obj):
_type = obj[0]
if _type == b'M':
return marshal.loads(obj[1:])
elif _type == b'P':
return pickle.loads(obj[1:])
else:
raise ValueError("invalid sevialization type: %s" % _type)
def loads(self, obj):
return self.serializer.loads(zlib.decompress(obj))
def loads(self, stream):
length = read_int(stream)
if length == SpecialLengths.END_OF_DATA_SECTION:
raise EOFError
elif length == SpecialLengths.NULL:
return None
s = stream.read(length)
return s.decode("utf-8") if self.use_unicode else s
def load_stream(self, stream):
try:
while True:
yield self.loads(stream)
except struct.error:
return
except EOFError:
return
def func_load(code, defaults=None, closure=None, globs=None):
'''Deserialize user defined function.'''
if isinstance(code, (tuple, list)): # unpack previous dump
code, defaults, closure = code
code = marshal.loads(code.encode('raw_unicode_escape'))
if closure is not None:
closure = func_reconstruct_closure(closure)
if globs is None:
globs = globals()
return python_types.FunctionType(code, globs, name=code.co_name, argdefs=defaults, closure=closure)
def read_pyc(filename):
"""
Read the code inside a bytecode compiled file if the MAGIC number is
compatible
Returns:
a code object
"""
data = read_file(filename, 'rb')
if not is_gae and data[:4] != imp.get_magic():
raise SystemError('compiled code is incompatible')
return marshal.loads(data[marshal_header_size:])