File: django_cache_base.py

package info (click to toggle)
strawberry-graphql-django 0.62.0-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 1,968 kB
  • sloc: python: 27,530; sh: 17; makefile: 16
file content (54 lines) | stat: -rw-r--r-- 1,998 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
from collections.abc import Callable, Hashable
from functools import _make_key  # noqa: PLC2701
from typing import Optional, cast

from django.core.cache import caches
from django.core.cache.backends.base import DEFAULT_TIMEOUT
from strawberry.extensions import SchemaExtension
from strawberry.types import ExecutionContext


class DjangoCacheBase(SchemaExtension):
    """Base for a Cache that uses Django built in cache instead of an in memory cache.

    Arguments:
    ---------
    `cache_name: str`
        Name of the Django Cache to use, defaults to 'default'

    `timeout: Optional[int]`
        How long to hold items in the cache. See the Django Cache docs for details
        https://docs.djangoproject.com/en/4.0/topics/cache/

    `hash_fn: Optional[Callable[[Tuple, Dict], str]]`
        A function to use to generate the cache keys
        Defaults to the same key generator as functools.lru_cache
        WARNING! The default function does NOT work with memcached
        and will generate warnings

    """

    def __init__(
        self,
        cache_name: str = "default",
        timeout: Optional[int] = None,
        hash_fn: Optional[Callable[[tuple, dict], Hashable]] = None,
        *,
        execution_context: Optional[ExecutionContext] = None,
    ):
        super().__init__(execution_context=cast("ExecutionContext", execution_context))

        self.cache = caches[cache_name]
        self.timeout = timeout or DEFAULT_TIMEOUT
        # Use same key generating function as functools.lru_cache as default
        self.hash_fn = hash_fn or (lambda args, kwargs: _make_key(args, kwargs, False))

    def execute_cached(self, func, *args, **kwargs):
        hash_key = cast("str", self.hash_fn(args, kwargs))
        cache_result = self.cache.get(hash_key)
        if cache_result is not None:
            return cache_result

        func_result = func(*args, **kwargs)
        self.cache.set(hash_key, func_result, timeout=self.timeout)
        return func_result