Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def make_client(self, *mock_socket_values, **kwargs):
current_port = TEST_PORT
from pymemcache.client.hash import HashClient
self.client = HashClient([], **kwargs)
ip = TEST_HOST
for vals in mock_socket_values:
s = '{}:{}'.format(ip, current_port)
c = self.make_client_pool((ip, current_port), vals, **kwargs)
self.client.clients[s] = c
self.client.hasher.add_node(s)
current_port += 1
return self.client
- python-memcached: ['127.0.0.1:11211', ...] or ['127.0.0.1', ...]
- pymemcache: [('127.0.0.1', 11211), ...]
"""
_host_and_port_list = []
for server in servers:
connection_info = server.split(':')
if len(connection_info) == 1:
_host_and_port_list.append(
(connection_info[0], 11211))
elif len(connection_info) == 2:
_host_and_port_list.append(
(connection_info[0], int(connection_info[1])))
return _host_and_port_list
class Client(HashClient):
"""pymemcache client.
Customize pymemcache behavior as python-memcached (default backend)'s one.
"""
def __init__(self, servers, *args, **kwargs):
kwargs['serializer'] = python_memcache_serializer
kwargs['deserializer'] = python_memcache_deserializer
return super(Client, self).__init__(
_split_host_and_port(servers), *args, **kwargs)
def disconnect_all(self):
for client in self.clients.values():
client.close()
def get_many(self, keys, gets=False, *args, **kwargs):
nodes is returned, otherwise the current client is returned.
"""
# Collect the all Memcached pods' IP addresses
try:
_, _, ips = socket.gethostbyname_ex(self.host)
except socket.gaierror:
# The host could not be found. This mean that either the service is
# down or that no pods are running
ips = []
if set(ips) != set(self._ips):
# A different list of ips has been detected, so we generate
# a new client
self._ips = ips
if self._ips:
servers = [(ip, self.port) for ip in self._ips]
self._client = HashClient(servers, use_pooling=True)
else:
self._client = None
def connect(self, server):
server = CacheMemcacheStrategy.splitHosts(server)
assert server, "{} is not a suitable server".format(server)
if len(server) == 1:
clientClass = Client
server = server[0]
else:
from pymemcache.client.hash import HashClient
clientClass = HashClient
self.client = clientClass(server, ignore_exc=True,
serializer=python_memcache_serializer,
deserializer=python_memcache_deserializer,
timeout=5,
connect_timeout=5,
key_prefix=(getStringHash(self.fileStrategy.dir) + "_").encode("UTF-8")
)