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})
python类get_or_set()的实例源码
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
