Python django.core.cache.cache 模块，set_many() 实例源码

def test_binary_string(self):
        # Binary strings should be cacheable
        cache = self.cache
        from zlib import compress, decompress
        value = 'value_to_be_compressed'
        compressed_value = compress(value.encode())

        # Test set
        cache.set('binary1', compressed_value)
        compressed_result = cache.get('binary1')
        self.assertEqual(compressed_value, compressed_result)
        self.assertEqual(value, decompress(compressed_result).decode())

        # Test add
        cache.add('binary1-add', compressed_value)
        compressed_result = cache.get('binary1-add')
        self.assertEqual(compressed_value, compressed_result)
        self.assertEqual(value, decompress(compressed_result).decode())

        # Test set_many
        cache.set_many({'binary1-set_many': compressed_value})
        compressed_result = cache.get('binary1-set_many')
        self.assertEqual(compressed_value, compressed_result)
        self.assertEqual(value, decompress(compressed_result).decode())

def test_long_timeout(self):
        """
        Followe memcached's convention where a timeout greater than 30 days is
        treated as an absolute expiration timestamp instead of a relative
        offset (#12399).
        """
        cache = self.cache
        cache.set('key1', 'eggs', 60 * 60 * 24 * 30 + 1)  # 30 days + 1 second
        self.assertEqual(cache.get('key1'), 'eggs')

        cache.add('key2', 'ham', 60 * 60 * 24 * 30 + 1)
        self.assertEqual(cache.get('key2'), 'ham')

        cache.set_many({'key3': 'sausage', 'key4': 'lobster bisque'}, 60 * 60 * 24 * 30 + 1)
        self.assertEqual(cache.get('key3'), 'sausage')
        self.assertEqual(cache.get('key4'), 'lobster bisque')

def get_all_contest_participants_detail(contest: Contest, users=None, privilege=False):
    cache_template = PARTICIPANT_RANK_DETAIL_PRIVATE if privilege else PARTICIPANT_RANK_DETAIL
    timeout = 60 if privilege else FORTNIGHT
    contest_users = users if users else contest.participants_ids
    cache_names = list(map(lambda x: cache_template.format(contest=contest.pk, user=x), contest_users))
    cache_res = cache.get_many(cache_names)
    ans = dict()
    second_attempt = []
    for user in contest_users:
        cache_name = cache_template.format(contest=contest.pk, user=user)
        if cache_name not in cache_res.keys():
            second_attempt.append(user)
        else:
            ans[user] = cache_res[cache_name]
    if second_attempt:
        ans2 = recalculate_for_participants(contest, second_attempt, privilege)
        cache.set_many({cache_template.format(contest=contest.pk, user=user_id): val for user_id, val in ans2.items()},
                       timeout * uniform(0.8, 1))
        ans.update(ans2)
    return ans

def test_unicode(self):
        # Unicode values can be cached
        cache = self.cache
        stuff = {
            'ascii': 'ascii_value',
            'unicode_ascii': 'Iñtërnâtiônàlizætiøn1',
            'Iñtërnâtiônàlizætiøn': 'Iñtërnâtiônàlizætiøn2',
            'ascii2': {'x': 1}
        }
        # Test `set`
        for (key, value) in stuff.items():
            cache.set(key, value)
            self.assertEqual(cache.get(key), value)

        # Test `add`
        for (key, value) in stuff.items():
            cache.delete(key)
            cache.add(key, value)
            self.assertEqual(cache.get(key), value)

        # Test `set_many`
        for (key, value) in stuff.items():
            cache.delete(key)
        cache.set_many(stuff)
        for (key, value) in stuff.items():
            self.assertEqual(cache.get(key), value)

def test_set_many(self):
        # Multiple keys can be set using set_many
        cache = self.cache
        cache.set_many({"key1": "spam", "key2": "eggs"})
        self.assertEqual(cache.get("key1"), "spam")
        self.assertEqual(cache.get("key2"), "eggs")

def test_set_many_expiration(self):
        # set_many takes a second ``timeout`` parameter
        cache = self.cache
        cache.set_many({"key1": "spam", "key2": "eggs"}, 1)
        time.sleep(2)
        self.assertIsNone(cache.get("key1"))
        self.assertIsNone(cache.get("key2"))

def test_zero_timeout(self):
        """
        Passing in zero into timeout results in a value that is not cached
        """
        cache = self.cache
        cache.set('key1', 'eggs', 0)
        self.assertIsNone(cache.get('key1'))

        cache.add('key2', 'ham', 0)
        self.assertIsNone(cache.get('key2'))

        cache.set_many({'key3': 'sausage', 'key4': 'lobster bisque'}, 0)
        self.assertIsNone(cache.get('key3'))
        self.assertIsNone(cache.get('key4'))

def global_template_vars(request):
    """
    Adds variables to all templates.

    It uses memcached to minimize hitting the db.
    """
    def get_user():
        if request.user.is_authenticated():
            return ExtendedUser.objects.get( id = request.user.id )
        else:
            return None
    vars_funcs = {
            'SITE_NAME':    lambda: Site.objects.get_current().name,
            'SITE_DOMAIN':  lambda: Site.objects.get_current().domain,
            'USERS_NR':     lambda: User.objects.count(),
            'EVENTS_NR':    lambda: Event.objects.count(),
            'GROUPS_NR':    lambda: Group.objects.count(), }
            # protocol  (computed below)
    vars_dic = cache.get_many( vars_funcs.keys() )
    if not vars_dic:
        vars_dic = {}
        # we get the values
        for key, func in vars_funcs.items():
            vars_dic[ key ] = func()
        # we put the values in the cache
        cache.set_many( vars_dic )
    # we add protocol
    if request.is_secure():
        vars_dic['PROTOCOL'] = "https"
    else:
        vars_dic['PROTOCOL'] = "http"
    vars_dic['VERSION'] = settings.VERSION
    vars_dic['MEDIA_URL'] = settings.MEDIA_URL
    # TODO: think on the trick to get the user out of a signed Django-1.4 cookie
    vars_dic['USER'] = get_user()
    vars_dic['READ_ONLY'] = settings.READ_ONLY
    # return
    return vars_dic

def test_set_get_many(self):
        data = {
            'key1': 'value1',
            'key2': 'value2'
        }
        await cache.set_many(data)
        results = await cache.get_many(data.keys())
        self.assertDictEqual(results, data)
        data['non_existing'] = None
        results = await cache.get_many(data.keys())
        self.assertDictEqual(results, data)

def get_cached_multiple_link_load(items, time_interval):
    """Cached version of get_multiple_link_load()"""
    item_map = {k: _cache_key(k, time_interval) for k in iterkeys(items)}
    # cache lookup
    cached = cache.get_many(item_map.values())
    _logger.debug(
        "get_cached_multiple_link_load: got %d/%d values from cache (%r)",
        len(cached), len(items), time_interval)

    # retrieve data for cache misses
    misses = {k: v
              for k, v in iteritems(items) if item_map[k] not in cached}
    if misses:
        get_multiple_link_load(misses, time_interval)

    # set data from cache
    reverse_item_map = {v: k for k, v in iteritems(item_map)}
    for cache_key, value in iteritems(cached):
        key = reverse_item_map[cache_key]
        properties = items[key]
        properties['load_in'], properties['load_out'] = value

    # add new data to cache
    missed_data = {item_map[key]: (properties['load_in'],
                                   properties['load_out'])
                   for key, properties in iteritems(misses)}
    _logger.debug("get_cached_multiple_link_load: caching %d values",
                  len(missed_data))
    cache.set_many(missed_data, CACHE_TIMEOUT)

def get_cached_multiple_cpu_load(items, time_interval):
    """Cached version of get_multiple_link_load()"""
    item_map = {k: _cache_key(k, time_interval) for k in iterkeys(items)}
    # cache lookup
    cached = cache.get_many(item_map.values())
    _logger.debug(
        "get_cached_multiple_cpu_load: got %d/%d values from cache (%r)",
        len(cached), len(items), time_interval)

    # retrieve data for cache misses
    misses = {k: v
              for k, v in iteritems(items) if item_map[k] not in cached}
    if misses:
        get_multiple_cpu_load(misses, time_interval)

    # set data from cache
    reverse_item_map = {v: k for k, v in iteritems(item_map)}
    for cache_key, value in iteritems(cached):
        key = reverse_item_map[cache_key]
        properties = items[key]
        properties['load'] = value

    # add new data to cache
    missed_data = {item_map[key]: properties['load']
                   for key, properties in iteritems(misses)}
    _logger.debug("get_cached_multiple_cpu_load: caching %d values",
                  len(missed_data))
    cache.set_many(missed_data, CACHE_TIMEOUT)

def get_contest_rank_list(contest: Contest, privilege=False):
    def _calculate():

        def find_key(tup):
            if contest.penalty_counts:
                return tup[1]['score'], -tup[1]['penalty']
            else:
                return tup[1]['score']

        items = sorted(get_all_contest_participants_detail(contest, privilege=privilege).items(),
                       key=find_key, reverse=True)
        ans = []  # ans = [(user_id, rank), ...]
        last_item = None
        for idx, item in enumerate(items, start=1):
            if last_item and find_key(item) == find_key(last_item):
                ans.append((item[0], ans[-1][1]))
            else:
                ans.append((item[0], idx))
            last_item = item
        return ans

    if not privilege:
        # Try to use cache
        cache_name = PARTICIPANT_RANK_LIST.format(contest=contest.pk)
        t = cache.get(cache_name)
        if t is None:
            t = _calculate()
            cache.set(cache_name, t, FORTNIGHT * uniform(0.6, 1))
            d = {PARTICIPANT_RANK.format(contest=contest.pk, user=user): rank for user, rank in t}
            cache.set_many(d, FORTNIGHT * uniform(0.6, 1))
        return t
    else:
        return _calculate()

def test_cache_versioning_get_set_many(self):
        cache = self.cache
        cache2 = LRUObjectCache('lru2', dict(VERSION=2))
        cache2._cache = cache._cache

        # set, using default version = 1
        cache.set_many({'ford1': 37, 'arthur1': 42})
        self.assertEqual(cache.get_many(['ford1', 'arthur1']), {'ford1': 37, 'arthur1': 42})
        self.assertEqual(cache.get_many(['ford1', 'arthur1'], version=1), {'ford1': 37, 'arthur1': 42})
        self.assertEqual(cache.get_many(['ford1', 'arthur1'], version=2), {})

        self.assertEqual(cache2.get_many(['ford1', 'arthur1']), {})
        self.assertEqual(cache2.get_many(['ford1', 'arthur1'], version=1), {'ford1': 37, 'arthur1': 42})
        self.assertEqual(cache2.get_many(['ford1', 'arthur1'], version=2), {})

        # set, default version = 1, but manually override version = 2
        cache.set_many({'ford2': 37, 'arthur2': 42}, version=2)
        self.assertEqual(cache.get_many(['ford2', 'arthur2']), {})
        self.assertEqual(cache.get_many(['ford2', 'arthur2'], version=1), {})
        self.assertEqual(cache.get_many(['ford2', 'arthur2'], version=2), {'ford2': 37, 'arthur2': 42})

        self.assertEqual(cache2.get_many(['ford2', 'arthur2']), {'ford2': 37, 'arthur2': 42})
        self.assertEqual(cache2.get_many(['ford2', 'arthur2'], version=1), {})
        self.assertEqual(cache2.get_many(['ford2', 'arthur2'], version=2), {'ford2': 37, 'arthur2': 42})

        # v2 set, using default version = 2
        cache2.set_many({'ford3': 37, 'arthur3': 42})
        self.assertEqual(cache.get_many(['ford3', 'arthur3']), {})
        self.assertEqual(cache.get_many(['ford3', 'arthur3'], version=1), {})
        self.assertEqual(cache.get_many(['ford3', 'arthur3'], version=2), {'ford3': 37, 'arthur3': 42})

        self.assertEqual(cache2.get_many(['ford3', 'arthur3']), {'ford3': 37, 'arthur3': 42})
        self.assertEqual(cache2.get_many(['ford3', 'arthur3'], version=1), {})
        self.assertEqual(cache2.get_many(['ford3', 'arthur3'], version=2), {'ford3': 37, 'arthur3': 42})

        # v2 set, default version = 2, but manually override version = 1
        cache2.set_many({'ford4': 37, 'arthur4': 42}, version=1)
        self.assertEqual(cache.get_many(['ford4', 'arthur4']), {'ford4': 37, 'arthur4': 42})
        self.assertEqual(cache.get_many(['ford4', 'arthur4'], version=1), {'ford4': 37, 'arthur4': 42})
        self.assertEqual(cache.get_many(['ford4', 'arthur4'], version=2), {})

        self.assertEqual(cache2.get_many(['ford4', 'arthur4']), {})
        self.assertEqual(cache2.get_many(['ford4', 'arthur4'], version=1), {'ford4': 37, 'arthur4': 42})
        self.assertEqual(cache2.get_many(['ford4', 'arthur4'], version=2), {})