Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# must be able to reliably map responses back to the original order
keys.append(key)
key = self.check_key(key)
data, data_flags = self.serde.serialize(key, data)
# If 'flags' was explicitly provided, it overrides the value
# returned by the serializer.
if flags is not None:
data_flags = flags
if not isinstance(data, six.binary_type):
try:
data = six.text_type(data).encode(self.encoding)
except UnicodeEncodeError as e:
raise MemcacheIllegalInputError(
"Data values must be binary-safe: %s" % e)
cmds.append(name + b' ' + key + b' ' +
six.text_type(data_flags).encode(self.encoding) +
b' ' + expire +
b' ' + six.text_type(len(data)).encode(self.encoding) +
extra + b'\r\n' + data + b'\r\n')
if self.sock is None:
self._connect()
try:
self.sock.sendall(b''.join(cmds))
if noreply:
return {k: True for k in keys}
try:
span.set_tags(p.tags)
vals = _get_query_string(args)
query = '{}{}{}'.format(method_name, ' ' if vals else '', vals)
span.set_tag(memcachedx.QUERY, query)
except Exception:
log.debug('Error setting relevant pymemcache tags')
try:
return method(*args, **kwargs)
except (
MemcacheClientError,
MemcacheServerError,
MemcacheUnknownCommandError,
MemcacheUnknownError,
MemcacheIllegalInputError,
):
(typ, val, tb) = sys.exc_info()
span.set_exc_info(typ, val, tb)
reraise(typ, val, tb)
def _check_integer(self, value, name):
"""Check that a value is an integer and encode it as a binary string"""
if not isinstance(value, six.integer_types): # includes "long" on py2
raise MemcacheIllegalInputError(
'%s must be integer, got bad value: %r' % (name, value)
)
return six.text_type(value).encode(self.encoding)