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_pool(self, hostname, mock_socket_values, serializer=None, **kwargs):
mock_client = pymemcache.client.base.Client(
hostname, serializer=serializer, **kwargs
)
tracer = get_dummy_tracer()
Pin.override(mock_client, tracer=tracer)
mock_client.sock = MockSocket(mock_socket_values)
client = pymemcache.client.base.PooledClient(hostname, serializer=serializer)
client.client_pool = pymemcache.pool.ObjectPool(lambda: mock_client)
return mock_client
def setUp(self):
pool = PooledClient(server=memcached_endpoint.address)
self.backend_factory = MemcacheRateLimitBackendContextFactory(pool)
super().setUp()
def add_server(self, server, port):
key = '%s:%s' % (server, port)
if self.use_pooling:
client = PooledClient(
(server, port),
**self.default_kwargs
)
else:
client = Client((server, port), **self.default_kwargs)
self.clients[key] = client
self.hasher.add_node(key)
:returns: :py:class:`pymemcache.client.base.PooledClient`
"""
assert prefix.endswith(".")
parser = config.SpecParser(
{
"endpoint": config.Endpoint,
"max_pool_size": config.Optional(config.Integer, default=None),
"connect_timeout": config.Optional(config.TimespanWithLegacyFallback, default=None),
"timeout": config.Optional(config.TimespanWithLegacyFallback, default=None),
"no_delay": config.Optional(config.Boolean, default=True),
}
)
options = parser.parse(prefix[:-1], app_config)
return PooledClient(
server=options.endpoint.address,
connect_timeout=options.connect_timeout and options.connect_timeout.total_seconds(),
timeout=options.timeout and options.timeout.total_seconds(),
serializer=serializer,
deserializer=deserializer,
no_delay=options.no_delay,
max_pool_size=options.max_pool_size,
)