'''INTERNAL: Python 2 does not define this exception, so we

'''INTERNAL: Convert bytes to a regular string.'''

if PY3:

if isinstance(data, str):

return data

return str(data, 'latin-1')

'''INTERNAL: Convert something to bytes type.'''

if PY3:

if isinstance(data, bytes):

return data

return bytes(data, 'latin-1')

'''INTERNAL: Convert a stats boolean string into boolean.'''

if s in ['0', 'no']:

return False

if s in ['1', 'yes']:

return True

buffers_by_server, send_threshold, send_minimum,

expected_keys, results, return_successful, return_failed):

'''INTERNAL: Write and read to sockets that are ready.

return_exception = None

read_sockets = [x for x in buffers_by_server.keys() if x.backend]

write_sockets = [

x[0] for x in buffers_by_server.items()

if not send_threshold or len(x[1]) >= send_threshold]

read_ready, write_ready = select.select(

read_sockets, write_sockets, [])[:2]

# receive data from read-ready sockets

for server in read_ready:

try:

server.read_from_socket()

except ServerDisconnect:

return_exception = _deferred_exception(

server, results, buffers_by_server, expected_keys,

'ServerDisconnect received, probably server died')

while server.line_available() and expected_keys[server]:

line = server.read_until().rstrip()

key = expected_keys[server].pop(0)

if line.startswith('CLIENT_ERROR'):

return_exception = _deferred_exception(

server, results, buffers_by_server, expected_keys,

'CLIENT_ERROR received, possibly key is too long.')

if line == 'STORED':

if return_successful:

results[key] = None

elif return_failed:

results[key] = line

# send data to write-ready sockets

for server in write_ready:

data = buffers_by_server[server]

if not data:

continue

bytes_sent = server.backend.send(data)

del data[:bytes_sent]

server, results, buffers_by_server, expected_keys, message):

'''INTERNAL: Handle an exception on a socket during set_multi.

server.reset()

server.connect()

del buffers_by_server[server][:]

del expected_keys[server][:]

'''INTERNAL: Write a debugging message to stderr with a stack trace.

import inspect

import os

stack = (

' -> '.join(['{0}({1}:{2})'.format(x[3],

os.path.basename(x[1]), x[2])

for x in reversed(inspect.stack()[1:])]))

sys.stderr.write('{0} => {1}\n'.format(msg, stack))

'''There are no servers available to cache on, probably because all

'''Item was not stored, but not due to an error. Normally means the

'''Item you are trying to store with a "cas" command has been modified

'''During a SET operation the server returned CLIENT_ERROR. This is

def __init__(self, message=None, results={}):

self.message = message

'''When trying to refresh a CAS from the memcached, the retrieved value

'''A dictionary-like interface which swallows server exceptions.

def __init__(self, servers, selector=None, hasher=None):

ret = super(ExceptionsAreMissesMapping, self).__init__()

self.memcache = Memcache(servers, selector, hasher)

return ret

def __getitem__(self, key):

try:

return self.memcache.get(key)

except (NoValue, ServerDisconnect, NotStored, NotFound, CASFailure):

raise KeyError(key)

def __setitem__(self, key, value):

try:

self.memcache.set(key, value)

except (ServerDisconnect, NotStored, NotFound, CASFailure):

pass

def __delitem__(self, key):

try:

self.memcache.delete(key)

return True

except (ServerDisconnect, NoAvailableServers, NotFound):

return False

def __iter__(self):

raise NotImplementedError()

def __len__(self):

items = 0

try:

for server_stats in self.memcache.stats():

items += server_stats.get('curr_items', 0)

except (ServerDisconnect, NoAvailableServers):

pass

'''Wrapper around Memcache value results.

def __new__(self, value, key, flags, cas_unique=None, memcache=None):

'''Instantiate new instance.

data = super(ValueSuperStr, self).__new__(self, value)

data.key = key

data.flags = flags

data.cas_unique = cas_unique

data.cas_unavailable = False

data.memcache = memcache

return data

def set(self, value, flags=0, exptime=0, update_cas=False):

'''Update the value in the server.

if self.cas_unique:

if self.cas_unavailable:

raise CASFailure('CAS value already consumed.')

self.cas_unavailable = True

retval = self.memcache.set(

self.key, value, flags=flags, exptime=exptime,

cas_unique=self.cas_unique)

if self.cas_unique and update_cas:

result = self.memcache.get(self.key)

if result != value:

raise CASRefreshFailure(

'Value from server changed during CAS refresh')

self.cas_unique = result.cas_unique

self.cas_unavailable = False

return retval

def append(self, value):

'''Append `value` to the data stored for this key.

if self.cas_unique is not None:

raise CASFailure('Not supported with CAS')

return self.memcache.append(self.key, value)

def prepend(self, value):

'''Prepend `value` to the data stored for this key.

if self.cas_unique is not None:

raise CASFailure('Not supported with CAS')

return self.memcache.prepend(self.key, value)

def incr(self, value=1):

'''Increment the value for this key.

if self.cas_unique is not None:

raise CASFailure('Not supported with CAS')

return self.memcache.incr(self.key, value)

def decr(self, value=1):

'''Decrement the value for this key.

if self.cas_unique is not None:

raise CASFailure('Not supported with CAS')

return self.memcache.decr(self.key, value)

def delete(self):

'''Remove this key from the server.

if self.cas_unique is not None:

raise CASFailure('Not supported with CAS')

return self.memcache.delete(self.key)

def delete_all(self):

'''Remove this key from all of the the servers.

if self.cas_unique is not None:

raise CASFailure('Not supported with CAS')

return self.memcache.delete_all(self.key)

def touch(self, exptime):

'''Update the expiration time on an item.

if self.cas_unique is not None:

raise CASFailure('Not supported with CAS')

'''Encode the response as a dictionary.

def __init__(self, value, key, flags, cas_unique=None, memcache=None):

'''Instantiate new instance.

super(ValueDictionary, self).__init__(

[

['key', key],

['value', value],

['flags', flags],

['cas_unique', cas_unique]

'''Turn memcache keys into hashes, for use in server selection.

def hash(self, key):

'''Hash a key into a number.

'''Hasher that always returns 0, useful only for

def hash(self, key):

'''See :py:func:`memcached2.HasherBase.hash` for details of

'''Hasher compatible with the C memcache hash function.'''

def hash(self, key):

'''See :py:func:`memcached2.HasherBase.hash` for details of

key = _to_bytes(key)

'''A pool of servers that connections can be checked in/out from.

def __init__(self, reconnector=None):

'''Initialize ServerPool instance.

self.server_pools = {}

if reconnector is None:

reconnector = ReconnectorTime()

self.reconnector = reconnector

def _add(self, server_uri):

'''INTERNAL: Add the specified server to the pool list.

if server_uri in self.server_pools:

return

self.server_pools[server_uri] = queue.Queue()

def is_available(self, server_uri):

'''Is the server available for use. This checks internal state to

self._add(server_uri)

if not self.server_pools[server_uri].empty():

return True

return self.reconnector.connectable(server_uri)

def get(self, server_uri):

'''Retrieve a server for use. Either pulling it from the queue or

self._add(server_uri)

pool = self.server_pools[server_uri]

if not pool.empty():

try:

return pool.get_nowait()

except queue.Empty:

pass

server = ServerConnection(server_uri)

return server

def put(self, connection):

'''Return a connection to the pool.

if connection.buffer:

raise ValueError('Socket returned to pool with data waiting')

server_uri = connection.uri

self._add(server_uri)

self.server_pools[server_uri].put(connection)

def empty(self):

'''Release all the server connections currently in the pool.'''

for pool in self.server_pools.values():

if pool.empty():

continue

try:

connection = pool.get_nowait()

connection.reset()

except queue.Empty:

pass

def __del__(self):

'''Select which server to use.

def select(self, server_uri_list, hasher, key, server_pool):

'''Select a server bashed on the `key_hash`.

'''Server selector that only returns the first server. Useful when there

def select(self, server_uri_list, hasher, key, server_pool):

'''See :py:func:`memcached2.SelectorBase.select` for details of

server_uri = server_uri_list[0]

if server_pool.is_available(server_uri):

return server_uri

'''Select a server, if it is down re-hash up to `hashing_retries` times.

def __init__(self, hashing_retries=10):

'''

self.hashing_retries = hashing_retries

def select(self, server_uri_list, hasher, key, server_pool):

'''See :py:func:`memcached2.SelectorBase.select` for details of

server_uri = server_uri_list[hasher(key) % len(server_uri_list)]

if server_pool.is_available(server_uri):

return server_uri

for i in range(self.hashing_retries):

server_uri = server_uri_list[

hasher(key + str(i)) % len(server_uri_list)]

if server_pool.is_available(server_uri):

return server_uri

'''On a down server, re-partition that servers keyspace to other servers.

def select(self, server_uri_list, hasher, key, server_pool):

'''See :py:func:`memcached2.SelectorBase.select` for details of

key_hash = hasher(key)

remaining_uris = server_uri_list[:]

for i in range(len(server_uri_list)):

position = key_hash % len(remaining_uris)

server_uri = server_uri_list[position]

if server_pool.is_available(server_uri):

return server_uri

del(remaining_uris[position])

'''Predictably select a server, even if its normal server is down.

def __init__(self, total_buckets=None):

'''

self.total_buckets = None

self.buckets = None

def _initialize_buckets(self, server_uri_list, hasher):

'''Create the consistent-hashing set of buckets, used for determining

if self.buckets is not None:

return

len_uri_list = len(server_uri_list)

if not self.total_buckets:

self.total_buckets = len_uri_list * len_uri_list

bucket_dict = {}

for i in range(self.total_buckets):

bucket_dict[hasher(str(i))] = i % len_uri_list

self.buckets = sorted(bucket_dict.items())

def select(self, server_uri_list, hasher, key, server_pool):

'''See :py:func:`memcached2.SelectorBase.select` for details of

if self.buckets is None:

self._initialize_buckets(server_uri_list, hasher)

hashed_key = hasher(key)

offset = bisect(self.buckets, (hashed_key, 0))

len_buckets = len(self.buckets)

len_uri_list = len(server_uri_list)

already_tried_uris = set()

for i in range(len_buckets):

if len(already_tried_uris) == len_uri_list:

raise NoAvailableServers()

bucket_offset = (offset + i) % len_buckets

# the last bucket covers things up to the first bucket

if bucket_offset == 0 and hashed_key < self.buckets[0][0]:

bucket = self.buckets[-1]

else:

bucket = self.buckets[bucket_offset]

uri_offset = bucket[1]

server_uri = server_uri_list[uri_offset]

if uri_offset in already_tried_uris:

continue

already_tried_uris.update([uri_offset])

if server_pool.is_available(server_uri):

return server_uri

'''Track server problems and determine when to reconnect.

def __init__(self):

pass

def connectable(self, server_url):

'''Is the specified server connectable?

raise NotImplementedError()

def had_error(self, server_url):

'''Called to report to the reconnector when there is an error on

raise NotImplementedError()

def had_success(self, server_url):

'''Called when a successful communication occurs with a server.

'''This is a simple reconnector that immediately tries connections on

def __init__(self):

ReconnectorBase.__init__(self)

self.server_data = {}

def connectable(self, server_url):

'''See :py:func:`memcached2.ReconnectorBase.connectable` for

return True

def had_error(self, server_url):

'''See :py:func:`memcached2.ReconnectorBase.error` for details of

pass

def had_success(self, server_url):

'''See :py:func:`memcached2.ReconnectorBase.had_success` for

'''Try reconnecting to a server after a specific number of seconds

def __init__(self, timeout=30):

'''

ReconnectorBase.__init__(self)

self.timeout = timeout

self.last_error = {}

def connectable(self, server_url):

'''See :py:func:`memcached2.ReconnectorBase.connectable` for

last_error = self.last_error.get(server_url)

if last_error is None:

return True

now = time.time()

if now - last_error >= self.timeout:

self.had_success(server_url)

return True

return False