我们从Python开源项目中，提取了以下12个代码示例，用于说明如何使用cPickle.PicklingError()。
def test_reduce_bad_iterator(self): # Issue4176: crash when 4th and 5th items of __reduce__() # are not iterators class C(object): def __reduce__(self): # 4th item is not an iterator return list, (), None, [], None class D(object): def __reduce__(self): # 5th item is not an iterator return dict, (), None, None, [] # Protocol 0 in Python implementation is less strict and also accepts # iterables. for proto in protocols: try: self.dumps(C(), proto) except (AttributeError, pickle.PicklingError, cPickle.PicklingError): pass try: self.dumps(D(), proto) except (AttributeError, pickle.PicklingError, cPickle.PicklingError): pass
def putmessage(self, message): self.debug("putmessage:%d:" % message[0]) try: s = pickle.dumps(message) except pickle.PicklingError: print >>sys.__stderr__, "Cannot pickle:", repr(message) raise s = struct.pack("<i", len(s)) + s while len(s) > 0: try: r, w, x = select.select([], [self.sock], []) n = self.sock.send(s[:BUFSIZE]) except (AttributeError, TypeError): raise IOError, "socket no longer exists" except socket.error: raise else: s = s[n:]
def put(self, o): """Encode object ``o`` and write it to the pipe. Block gevent-cooperatively until all data is written. The default encoder is ``pickle.dumps``. :arg o: a Python object that is encodable with the encoder of choice. Raises: - :exc:`GIPCError` - :exc:`GIPCClosed` - :exc:`pickle.PicklingError` """ self._validate() with self._lock: bindata = self._encoder(o) self._write(struct.pack("!i", len(bindata)) + bindata)
def dump(self, value, f): try: pickle.dump(value, f, 2) except pickle.PickleError: raise except Exception as e: msg = "Could not serialize broadcast: " + e.__class__.__name__ + ": " + e.message print_exec(sys.stderr) raise pickle.PicklingError(msg) f.close() return f.name
def pickled(cls, obj, **kwargs): self = cls(**kwargs) try: self.data = cPickle.dumps(obj, protocol=2) except cPickle.PicklingError, e: self.data = cPickle.dumps(CallError(e), protocol=2) return self
def putmessage(self, message): self.debug("putmessage:%d:" % message[0]) try: s = pickle.dumps(message) except pickle.PicklingError: print >>sys.__stderr__, "Cannot pickle:", repr(message) raise s = struct.pack("<i", len(s)) + s while len(s) > 0: try: r, w, x = select.select([], [self.sock], []) n = self.sock.send(s[:BUFSIZE]) except (AttributeError, TypeError): raise IOError, "socket no longer exists" s = s[n:]
def update_saved_parameters(training_stats, test_stack_config, args): if training_stats['worthy_epoch']: test_stack_config['best_epoch_id'] = training_stats['best_epoch_id'] test_stack_config['validation_result'] = training_stats['validation_result'] test_stack_config['training_result'] = training_stats['training_result'] try: lstm_seqlabel_load_save_model.save_parameters_to_file( test_stack_config, args.saveto) except pickle.PicklingError as e: print "Suffering from Pickling Error in saving \n%s \nto %s"%( ' '.join(test_stack_config.keys()), args.saveto) return
def test_handledByPickleModule(self): """ Handling L{pickle.PicklingError} handles L{_UniversalPicklingError}. """ self.assertRaises(pickle.PicklingError, self.raise_UniversalPicklingError)
def test_handledBycPickleModule(self): """ Handling L{cPickle.PicklingError} handles L{_UniversalPicklingError}. """ try: import cPickle except ImportError: raise unittest.SkipTest("cPickle not available.") else: self.assertRaises(cPickle.PicklingError, self.raise_UniversalPicklingError)
def test_lambdaRaisesPicklingError(self): """ Pickling a C{lambda} function ought to raise a L{pickle.PicklingError}. """ self.assertRaises(pickle.PicklingError, pickle.dumps, lambdaExample) try: import cPickle except: pass else: self.assertRaises(cPickle.PicklingError, cPickle.dumps, lambdaExample)
