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

def about(request):
    """Information about the current site, its goals, and what content is loaded"""
    # Provider counts
    providers = cache.get_or_set(CACHE_STATS_NAME, [], CACHE_STATS_DURATION)
    if not providers:
        for provider in sorted(settings.PROVIDERS.keys()):
            s = Search()
            q = Q('term', provider=provider)
            s = s.query(q)
            response = s.execute()
            if response.hits.total > 0:
                data = settings.PROVIDERS[provider]
                total = intcomma(response.hits.total)
                data.update({'hits': total})
                providers.append(data)
        # All results
        s = Search()
        response = s.execute()
        total = intcomma(response.hits.total)
        providers.append({'display_name': 'Total', 'hits': total})
        cache.set(CACHE_STATS_NAME, providers)
    return render(request, "about.html", {'providers': providers})

def get_spec(name, http_client=None, config=None):
    """
    :param name: Name of the revision of spec, eg latest or v4
    :param http_client: Requests client used for retrieving specs
    :param config: Spec configuration - see Spec.CONFIG_DEFAULTS
    :return: :class:`bravado_core.spec.Spec`
    """
    http_client = http_client or requests_client.RequestsClient()

    def load_spec():
        loader = Loader(http_client)
        return loader.load_spec(build_spec_url(name))

    spec_dict = cache.get_or_set(build_cache_name(name), load_spec, app_settings.ESI_SPEC_CACHE_DURATION)
    config = dict(CONFIG_DEFAULTS, **(config or {}))
    return Spec.from_dict(spec_dict, build_spec_url(name), http_client, config)

def user_board_listing(request, account_id):
    """Return a paginated board listing view for account with account_id."""
    metric_timeout_seconds = 60 * 2

    queries = {
        # default to boards contributed to
        None: lambda x: ('contributed to', user_boards_contributed(x)),
        'created': lambda x: ('created', user_boards_created(x)),
        'evaluated': lambda x: ('evaluated', user_boards_evaluated(x)),
        'contribute': lambda x: ('contributed to', user_boards_contributed(x)),
    }

    user = get_object_or_404(User, pk=account_id)
    query = request.GET.get('query')
    verb, board_list = queries.get(query, queries[None])(user)
    desc = _('List of intelligence boards user {username} has {verb}').format(username=user.username, verb=verb)
    context = {
        'user': user,
        'boards': make_paginator(request, board_list),
        'contributors': cache.get_or_set('contributor_count', generate_contributor_count(), metric_timeout_seconds),
        'evaluators': cache.get_or_set('evaluator_count', generate_evaluator_count(), metric_timeout_seconds),
        'meta_description': desc,
        'verb': verb
    }
    return render(request, 'boards/user_boards.html', context)

def rejudge_submission_set(submission_set):
    cache.set(REJUDGE_COUNTER, 0)

    def decrease_counter():
        try:
            cache.decr(REJUDGE_COUNTER, 1)
        except ValueError:
            cache.set(REJUDGE_COUNTER, 0)

    with transaction.atomic():
        for submission in submission_set:
            submission.status_private = submission.status = SubmissionStatus.WAITING
            submission.status_detail = ""
            submission.status_message = ""
            submission.save(update_fields=["status_private", "status", "status_detail", "status_message"])

    for submission in submission_set:
        while cache.get_or_set(REJUDGE_COUNTER, 0) >= REJUDGE_TASK_LIMIT:
            sleep(1)
        try:
            cache.incr(REJUDGE_COUNTER, 1)
        except:
            cache.set(REJUDGE_COUNTER, 0)
        rejudge_submission(submission, decrease_counter)

def _retrieve_or_save(k, default_v):
    return cache.get_or_set(k, default_v, timeout=settings.ACTIVATION_LOGGING_CACHE_TIMEOUT)

def _append(list_key, v):
    the_list = cache.get_or_set(list_key, [], timeout=settings.ACTIVATION_LOGGING_CACHE_TIMEOUT)
    the_list.append(v)
    _save(list_key, the_list)

def _update_timestamp(k):
    cache.get_or_set(k, timezone.now, timeout=settings.ACTIVATION_LOGGING_CACHE_TIMEOUT)


# Public Methods

def list(self, request, *args, **kwargs):
        """
        Returns a list of all currently available projects with links to
        relevant API endpoints
        """
        # Project list can only change with code addition and app restart,
        # so it makes sense to cache it on first request after restart
        projects = cache.get_or_set(
            key=utils.AVAILABLE_PROJECTS_KEY,
            default=utils.get_projects(request),
            timeout=None,  # This will be invalidated on app restart
        )

        return Response(projects)

def retrieve(self, request, *args, **kwargs):
        """
        Returns project latest stable version
        """
        project = self.get_object()

        latest_version = cache.get_or_set(
            key=utils.get_latest_version_key(project.slug),
            default=project.get_latest_version,
            timeout=60 * 60,  # 1 hour
        )

        return Response({
            'latest': latest_version,
        })

def major(self, request, *args, **kwargs):
        """
        Returns project latest stable version for each major release
        """
        project = self.get_object()

        latest_versions = cache.get_or_set(
            key=utils.get_latest_major_versions_key(project.slug),
            default=project.get_latest_major_versions,
            timeout=60 * 60 * 6,  # 6 hours
        )

        return Response(latest_versions)

def minor(self, request,  *args, **kwargs):
        """
        Returns project latest stable version for each minor release
        """
        project = self.get_object()

        latest_versions = cache.get_or_set(
            key=utils.get_latest_minor_versions_key(project.slug),
            default=project.get_latest_minor_versions,
            timeout=60 * 60 * 6,  # 6 hours
        )

        return Response(latest_versions)

def board_listing(request):
    """Return a paginated board listing view showing all boards and their popularity."""
    board_list = Board.objects.order_by('-pub_date')
    metric_timeout_seconds = 60 * 2
    desc = _('List of intelligence boards on {name} and summary information').format(name=get_current_site(request).name)  # nopep8
    context = {
        'boards': make_paginator(request, board_list),
        'contributors': cache.get_or_set('contributor_count', generate_contributor_count(), metric_timeout_seconds),
        'evaluators': cache.get_or_set('evaluator_count', generate_evaluator_count(), metric_timeout_seconds),
        'meta_description': desc,
    }
    return render(request, 'boards/boards.html', context)

def _cache(self) -> Dict[str, Any]:
        if self._cached_obj is None:
            self._cached_obj = cache.get_or_set(
                'hierarkey_{}_{}'.format(self._cache_namespace, self._obj.pk),
                lambda: {s.key: s.value for s in self._objects.all()},
                timeout=1800
            )
        return self._cached_obj

def get_context_data(self, **kwargs):
        context = super(LeaderboardView, self).get_context_data(**kwargs)
        leaderboard = cache.get_or_set('leaderboard', self.calculate_leaderboard)
        context.update(leaderboard)
        return context

def cache_using_pk(func):
    """
    Given a model instance, cache the value from an instance method using the primary key
    """

    @functools.wraps(func)
    def wrapper(instance, *args, **kwargs):
        cache_key = '{func_name}-{pk}'.format(func_name=func.func_name, pk=instance.pk)
        return cache.get_or_set(cache_key, functools.partial(func, instance, *args, **kwargs))

    return wrapper