Login

CacheInDictManager

Author:
LLyaudet
Posted:
May 13, 2022
Language:
Python
Version:
3.2
Score:
0 (after 0 ratings)

This manager use a local (in python dicts) cache for efficiency. It caches get requests and is better used with a context manager. I based my work on this previous snippet : https://djangosnippets.org/snippets/815/

  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
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
import logging

from django.db.models.manager import BaseManager
from django.db.models.query import QuerySet


logger = logging.getLogger("CacheInDictManager")


class CacheInDictQuerySet(QuerySet):
    def __init__(
        self,
        model=None,
        query=None,
        using=None,
        hints=None,
        manager=None,
        debug=None,
    ):
        super().__init__(model, query, using, hints)
        self.manager = manager
        self.debug = debug

    def get(self, *args, **kwargs):
        instance = self.manager.get_instance_from_cache(*args, **kwargs)
        if instance is not None:
            return instance
        instance = super().get(*args, **kwargs)
        if self.debug:
            logger.info(f"CacheInDictQuerySet.get() instance fetched:{instance}")
        self.manager.fill_cache_for_instance(instance)
        return instance


class CacheInDictManager(BaseManager.from_queryset(CacheInDictQuerySet)):
    """
    A manager that caches get() request with some keys.
    Each key can be made of one or more fields of the model.
    """

    def __init__(self, *args, **kwargs):
        additional_custom_cache_keys_definitions = kwargs.pop(
            "additional_custom_cache_keys_definitions", None
        )
        super().__init__(*args, **kwargs)

        self.custom_cache_keys_definitions = {
            # each cache key definition must be ordered lexicographically
            # example : ("aaa_id", "bbb_id", "bca_id"),
            "pk": ("pk",),
            "id": ("id",),
        }
        if additional_custom_cache_keys_definitions is not None:
            self.custom_cache_keys_definitions.update(
                additional_custom_cache_keys_definitions
            )
        self.custom_cache = {
            key_identifier: {}
            for key_identifier in self.custom_cache_keys_definitions.keys()
        }
        self.debug = False
        if self.debug:
            logger.info(
                f"CacheInDictManager created id:{id(self)}"
                f" custom_cache_keys_definitions:{self.custom_cache_keys_definitions}"
            )

    def get_queryset(self):
        """
        Return a new QuerySet object. Subclasses can override this method to
        customize the behavior of the Manager.
        """
        return self._queryset_class(
            model=self.model,
            using=self._db,
            hints=self._hints,
            manager=self,
            debug=self.debug,
        )

    @staticmethod
    def get_key_identifier(key):
        return "_".join(key)

    @staticmethod
    def get_key_value_for_instance(key_definition, instance):
        sub_key_values = []
        for sub_key in key_definition:
            if not hasattr(instance, sub_key):
                return None
            sub_key_value = getattr(instance, sub_key)
            if sub_key_value is None:
                return None
            if not isinstance(sub_key_value, int):
                return None
            if sub_key_value <= 0:
                return None
            sub_key_values.append(str(sub_key_value))
        return "_".join(sub_key_values)

    def fill_cache_for_instance(self, instance):
        if self.debug:
            logger.info(
                f"CacheInDictManager.fill_cache_for_instance() instance:{instance}"
            )
        for (
            key_identifier,
            key_definition,
        ) in self.custom_cache_keys_definitions.items():
            key_value = CacheInDictManager.get_key_value_for_instance(
                key_definition, instance
            )
            if self.debug:
                logger.info(
                    "CacheInDictManager.fill_cache_for_instance()"
                    f" key_identifier:{key_identifier} key_value:{key_value}"
                )
            self.custom_cache[key_identifier][key_value] = instance

    def get_instance_from_cache(self, *args, **kwargs):
        sorted_kwargs_keys = list(kwargs.keys())
        sorted_kwargs_keys.sort(key=lambda x: x)
        key_identifier = CacheInDictManager.get_key_identifier(sorted_kwargs_keys)
        if self.debug:
            logger.info(
                f"CacheInDictManager.get_instance_from_cache() computed key_identifier:{key_identifier}"
            )
        if self.custom_cache_keys_definitions.get(key_identifier) is not None:
            if self.debug:
                logger.info(
                    "CacheInDictManager.get_instance_from_cache() key_identifier exists"
                )
            sorted_kwargs = list(kwargs.items())
            sorted_kwargs.sort(key=lambda x: x[0])
            key_value = "_".join(map(lambda x: str(x[1]), sorted_kwargs))
            if self.debug:
                logger.info(
                    f"CacheInDictManager.get_instance_from_cache() computed key_value:{key_value}"
                )
            instance = self.custom_cache[key_identifier].get(key_value)
            if instance is not None:
                if self.debug:
                    logger.info(
                        "CacheInDictManager.get_instance_from_cache() cache hit"
                    )
                return instance
        return None

More like this

  1. Month / Year SelectDateWidget based on django SelectDateWidget by pierreben 2 months, 4 weeks ago
  2. Python Django CRUD Example Tutorial by tuts_station 3 months, 2 weeks ago
  3. Browser-native date input field by kytta 4 months, 4 weeks ago
  4. Generate and render HTML Table by LLyaudet 5 months, 1 week ago
  5. My firs Snippets by GutemaG 5 months, 1 week ago

Comments

Please login first before commenting.