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
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,
    ):
        super().__init__(model, query, using, hints)
        self.manager = manager

    def get(self, *args, **kwargs):
        return self.manager.get(*args, **kwargs)


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

    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,
        )

    @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(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() computed key_identifier:{key_identifier}"
            )
        if self.custom_cache_keys_definitions.get(key_identifier) is not None:
            if self.debug:
                logger.info("CacheInDictManager.get() 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() 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() cache hit")
                return instance
        instance = super().get(*args, **kwargs)
        if self.debug:
            logger.info(f"CacheInDictManager.get() instance fetched:{instance}")
        self.fill_cache_for_instance(instance)
        return instance

More like this

  1. LazyPrimaryKeyRelatedField by LLyaudet 2 days, 11 hours ago
  2. MYSQL Full Text Expression by Bidaya0 3 days, 12 hours ago
  3. Custom model manager chaining (Python 3 re-write) by Spotted1270 1 week, 2 days ago
  4. Django Standard API Response Middleware for DRF for modern frontend easy usage by Denactive 3 weeks, 4 days ago
  5. EnhancedQuerySet by LLyaudet 1 month, 1 week ago

Comments

Please login first before commenting.