Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
value = json.dumps(mod)
chunks = int(math.ceil(len(value) / float(20000)))
uwsgi.cache_set(key, repr(chunks), 0, cache_chunks)
for j in range(0, chunks, 1):
uwsgi.cache_set(key + '-{}'.format(j),
value[j * 20000: (j + 1) * 20000], 0,
cache_modules)
chunks = int(math.ceil(len(json.dumps(modules)) / float(64000)))
for i in range(0, chunks, 1):
uwsgi.cache_set('modules-data{}'.format(i),
json.dumps(modules)[i * 64000: (i + 1) * 64000],
0, main_cache)
LOGGER.info(
'all {} modules chunks are set in uwsgi cache'.format(chunks))
uwsgi.cache_set('chunks-modules', repr(chunks), 0, cache_chunks)
chunks = int(math.ceil(len(json.dumps(vendors)) / float(64000)))
for i in range(0, chunks, 1):
uwsgi.cache_set('vendors-data{}'.format(i),
json.dumps(vendors)[i * 64000: (i + 1) * 64000],
0, main_cache)
LOGGER.info(
'all {} vendors chunks are set in uwsgi cache'.format(chunks))
uwsgi.cache_set('chunks-vendor', repr(chunks), 0, cache_chunks)
if response != 'work':
LOGGER.error('Could not load or create cache')
sys.exit(500)
uwsgi.cache_update('initialized', 'True', 0, cache_chunks)
def test_big_random(self):
blob = self.rand_blob(100000)
self.assertTrue(uwsgi.cache_set('KEY', blob, 0, 'items_1_100000'))
get_blob = uwsgi.cache_get('KEY', 'items_1_100000')
self.assertEqual(blob, get_blob)
self.assertTrue(uwsgi.cache_del('KEY', 'items_1_100000'))
self.assertIsNone(uwsgi.cache_set('KEY', 'X' * 100001, 0, 'items_1_100000'))
self.assertTrue(uwsgi.cache_set('KEY', 'X' * 10000, 0, 'items_1_100000'))
def test_multi_delete(self):
for i in range(0, 100):
self.assertTrue(uwsgi.cache_set('key1', 'X' * 50, 0, 'items_4_10'))
self.assertTrue(uwsgi.cache_del('key1', 'items_4_10'))
for i in range(0, 100):
self.assertIsNone(uwsgi.cache_set('key1', 'X' * 51, 0, 'items_4_10'))
self.assertIsNone(uwsgi.cache_del('key1', 'items_4_10'))
for i in range(0, 100):
self.assertTrue(uwsgi.cache_set('key1', 'X' * 50, 0, 'items_4_10'))
self.assertTrue(uwsgi.cache_del('key1', 'items_4_10'))
t.daemon = True
t.start()
def gl_func():
while True:
gevent.sleep(2)
print "i am a greenltet running in worker %d" % uwsgi.worker_id()
@postfork
def spawn_greenlet():
gevent.spawn(gl_func)
print uwsgi.version
print uwsgi.workers()
try:
uwsgi.cache_set('foo', "Hello World from cache")
except:
pass
def application(env, start_response):
print env['wsgi.input'].read()
if uwsgi.loop == 'gevent':
gevent.sleep()
start_response('200 OK', [('Content-Type', 'text/html')])
yield "foobar<br>"
if uwsgi.loop == 'gevent':
gevent.sleep(3)
yield str(env['wsgi.input'].fileno())
yield "<h1>Hello World</h1>"
try:
yield uwsgi.cache_get('foo')
except:
pass
def test_non_bitmap(self):
self.assertTrue(uwsgi.cache_set('KEY', 'X' * 20, 0, 'items_non_bitmap'))
self.assertTrue(uwsgi.cache_del('KEY', 'items_non_bitmap'))
self.assertIsNone(uwsgi.cache_set('KEY', 'X' * 21, 0, 'items_non_bitmap'))
self.assertTrue(uwsgi.cache_set('KEY', 'X' * 20, 0, 'items_non_bitmap'))
def test_lru(self):
self.assertTrue(uwsgi.cache_set('KEY1', 'X' * 20, 0, 'items_lru'))
self.assertTrue(uwsgi.cache_set('KEY2', 'X' * 20, 0, 'items_lru'))
self.assertTrue(uwsgi.cache_set('KEY3', 'Y' * 20, 0, 'items_lru'))
self.assertIsNone(uwsgi.cache_get('KEY1', 'items_lru'))
uwsgi.cache_get('KEY3', 'items_lru')
for i in range(4, 100):
self.assertTrue(uwsgi.cache_set('KEY%d' % i, 'Y' * 20, 0, 'items_lru'))
self.assertIsNone(uwsgi.cache_get('KEY%d' % (i-2), 'items_lru'))
def load_uwsgi_cache(cache_chunks, main_cache, cache_modules, on_change):
response = 'work'
initialized = uwsgi.cache_get('initialized', cache_chunks)
LOGGER.debug('initialized {} on change {}'.format(initialized, on_change))
if initialized is None or initialized == 'False' or on_change:
uwsgi.cache_clear(cache_chunks)
uwsgi.cache_clear(main_cache)
uwsgi.cache_clear(cache_modules)
if cache_chunks == 'cache_chunks1':
# set active cache to 2 until we work on cache 1
uwsgi.cache_set('active_cache', '2', 0, 'cache_chunks1')
uwsgi.cache_set('initialized', 'False', 0, cache_chunks)
response, data = make_cache(application.credentials, response, cache_chunks, main_cache, is_uwsgi=application.is_uwsgi)
cat = \
json.JSONDecoder(object_pairs_hook=collections.OrderedDict) \
.decode(data)['yang-catalog:catalog']
modules = cat['modules']
if cat.get('vendors'):
vendors = cat['vendors']
else:
vendors = {}
if len(modules) != 0:
for i, mod in enumerate(modules['module']):
key = mod['name'] + '@' + mod['revision'] + '/' + mod[
'organization']
value = json.dumps(mod)
def test_too_much_items(self):
self.assertTrue(uwsgi.cache_set('key1', 'HELLO', 0, 'items_4_10'))
self.assertTrue(uwsgi.cache_set('key2', 'HELLO', 0, 'items_4_10'))
self.assertTrue(uwsgi.cache_set('key3', 'HELLO', 0, 'items_4_10'))
self.assertTrue(uwsgi.cache_set('key4', 'HELLO', 0, 'items_4_10'))
self.assertIsNone(uwsgi.cache_set('key5', 'HELLO', 0, 'items_4_10'))
cat = \
json.JSONDecoder(object_pairs_hook=collections.OrderedDict) \
.decode(data)['yang-catalog:catalog']
modules = cat['modules']
if cat.get('vendors'):
vendors = cat['vendors']
else:
vendors = {}
if len(modules) != 0:
for i, mod in enumerate(modules['module']):
key = mod['name'] + '@' + mod['revision'] + '/' + mod[
'organization']
value = json.dumps(mod)
chunks = int(math.ceil(len(value) / float(20000)))
uwsgi.cache_set(key, repr(chunks), 0, cache_chunks)
for j in range(0, chunks, 1):
uwsgi.cache_set(key + '-{}'.format(j),
value[j * 20000: (j + 1) * 20000], 0,
cache_modules)
chunks = int(math.ceil(len(json.dumps(modules)) / float(64000)))
for i in range(0, chunks, 1):
uwsgi.cache_set('modules-data{}'.format(i),
json.dumps(modules)[i * 64000: (i + 1) * 64000],
0, main_cache)
LOGGER.info(
'all {} modules chunks are set in uwsgi cache'.format(chunks))
uwsgi.cache_set('chunks-modules', repr(chunks), 0, cache_chunks)
chunks = int(math.ceil(len(json.dumps(vendors)) / float(64000)))
for i in range(0, chunks, 1):
def on_post(self, req, resp):
body = load_body(req)
if uwsgi.cache_exists('configuration'):
uwsgi.cache_update('configuration', str(body))
else:
uwsgi.cache_set('configuration', str(body))
resp.status = falcon.HTTP_202