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
- Template tag - list punctuation for a list of items by shapiromatron 10 months, 4 weeks ago
- JSONRequestMiddleware adds a .json() method to your HttpRequests by cdcarter 11 months ago
- Serializer factory with Django Rest Framework by julio 1 year, 5 months ago
- Image compression before saving the new model / work with JPG, PNG by Schleidens 1 year, 6 months ago
- Help text hyperlinks by sa2812 1 year, 7 months ago
Comments
Please login first before commenting.