Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_coderize(raw_coder):
assert raw_coder
assert isinstance(coderize(raw_coder), Coder)
tuple_coder = lambda x: x, lambda x: x # noqa
registry.register('tuple', tuple_coder)
class NewStaticCoder(Coder):
@staticmethod
def encode(d):
return d
@staticmethod
def decode(d):
return d
registry.register('new_static', NewStaticCoder)
registry.register('new_static_obj', NewStaticCoder())
class NewCoder(Coder):
def encode(self, x):
return x
def decode(self, x):
return x
registry.register('new_obj', NewCoder())
def test_coder_registry():
registry = Registry()
error_coder = None, None
registry.register('_error', error_coder)
assert registry.get('_error') == (None, None)
tuple_coder = lambda x: x, lambda x: x # noqa
registry.register('tuple', tuple_coder)
class NewStaticCoder(Coder):
@staticmethod
def encode(d):
return d
@staticmethod
def decode(d):
return d
registry.register('new_static', NewStaticCoder)
registry.register('new_static_obj', NewStaticCoder())
class NewCoder(Coder):
def encode(self, x):
return x
def decode(self, x):
#: Default coder.
#:
#: encode and decode functions bypass the given parameter.
bypass_coder = bypass, bypass
#: Pickle coder.
#:
#: encode is :func:`pickle.dumps` and decode is :func:`pickle.loads`.
#: :mod:`cpickle` will be automatically loaded for CPython2.
pickle_coder = pickle_mod.dumps, pickle_mod.loads
class JsonCoder(Coder):
"""JSON Coder.
When :mod:`ujson` package is installed, `ujson` is automatically selected;
Otherwise, :mod:`json` will be used.
"""
@staticmethod
def encode(data):
"""Dump data to JSON string and encode it to UTF-8 bytes"""
return json_mod.dumps(data).encode('utf-8')
@staticmethod
def decode(binary):
"""Decode UTF-8 bytes to JSON string and load it to object"""
return json_mod.loads(binary.decode('utf-8'))
def coderize(raw_coder):
if isinstance(raw_coder, Coder):
coder = raw_coder
else:
if isinstance(raw_coder, tuple):
coder = CoderTuple(*raw_coder)
elif hasattr(raw_coder, 'encode') and hasattr(raw_coder, 'decode'):
coder = CoderTuple(raw_coder.encode, raw_coder.decode)
else:
raise TypeError(
"The given coder is not a coder compatibile object or "
"not a registered name in coder registry")
return coder
"""
@abc.abstractmethod
def encode(self): # pragma: no cover
"""Abstract encode function. Children must implement this function."""
pass
@abc.abstractmethod
def decode(self): # pragma: no cover
"""Abstract decode function. Children must implement this function."""
pass
#: Coder-compatible tuple with encode and decode functions
CoderTuple = namedtuple('Coder', ['encode', 'decode'])
Coder.register(CoderTuple)
def coderize(raw_coder):
if isinstance(raw_coder, Coder):
coder = raw_coder
else:
if isinstance(raw_coder, tuple):
coder = CoderTuple(*raw_coder)
elif hasattr(raw_coder, 'encode') and hasattr(raw_coder, 'decode'):
coder = CoderTuple(raw_coder.encode, raw_coder.decode)
else:
raise TypeError(
"The given coder is not a coder compatibile object or "
"not a registered name in coder registry")
return coder