Skip to content

serializer_cache#

SerializerCacheBase#

Helper base class for using a cache backend during serialization.

Provides a dict-based short term cache that fetches its data from a Django cache backend and supports committing changes back.

For usage example, see SerializerCacheSerializer.

Source code in src/apps/cache/serializer_cache.py 
 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
class SerializerCacheBase:
    """Helper base class for using a cache backend during serialization.

    Provides a dict-based short term cache that fetches its data from a
    Django cache backend and supports committing changes back.

    For usage example, see SerializerCacheSerializer.
    """

    # Attributes required to be set by subclasses
    cache_name: str  # Name of Django cache to use
    cached_fields: set  # Which fields are cached
    modified_attr: str  # Model attribute of latest modification timestamp

    # Internal fields
    changed: set  # Changed values not yet in source_cache

    def __init__(self, initial_instances: List[models.Model] = [], autocommit=True):
        self.get_source_cache()
        if not self.cached_fields:
            raise ValueError("Missing cached_fields")
        if not self.modified_attr:
            raise ValueError("Missing modified_attr")

        # When autocommit is enabled (default), changes are committed to source_cache immediately
        self.autocommit = autocommit
        self.changed = set()
        self.values = {}
        self.fetch_from_source(initial_instances)

    @classmethod
    def get_source_cache(cls) -> BaseCache:
        """Source cache is determined dynamically so it can be changed in tests."""
        return caches[cls.cache_name]

    def fetch_from_source(self, instances: List[models.Model], include_newer=False):
        """Fetch cached data from source cache.

        Fetch cached values that match modification timestamp of corresponding instance.
        """
        cached_values = self.get_source_cache().get_many([instance.id for instance in instances])
        for instance in instances:
            if cached := cached_values.get(instance.id):
                modified = getattr(instance, self.modified_attr)
                if cached["_modified"] == modified:
                    self.values[instance.id] = cached
                if include_newer and cached["_modified"] > modified:
                    # Include cache entries that are newer than instance
                    self.values[instance.id] = cached

    def commit_changed_to_source(self):
        changed = {key: self.values[key] for key in self.changed}
        self.get_source_cache().set_many(changed)
        self.clear_changed()

    def clear(self):
        self.values.clear()
        self.clear_changed()

    def clear_changed(self):
        self.changed.clear()

    def get_changed(self):
        return {k: v for k, v in self.values.items() if k in self.changed}

    def get_value(self, instance: models.Model) -> Optional[dict]:
        value = self.values.get(instance.id)
        if value is None:
            return None
        return {k: v for k, v in value.items() if k in self.cached_fields}

    def get_value_context(self, instance: models.Model):
        value = self.values.get(instance.id)
        if value is None:
            return None
        return value.get("_context")

    def set_value(
        self,
        instance: models.Model,
        value: dict,
        value_context: Optional[dict] = None,
        only_if_modified=False,
    ):
        """Set per-instance value to cache.

        The in-memory value is not copied so care needs to be taken
        to avoid modifying it after using set_value.
        """
        modified = getattr(instance, self.modified_attr)

        if only_if_modified:
            entry = self.values.get(instance.id) or {}
            if entry_modified := entry.get("_modified"):
                if modified <= entry_modified:
                    return  # Instance is same or older than existing cached entry

        self.changed.add(instance.id)
        self.values[instance.id] = {k: v for k, v in value.items() if k in self.cached_fields}
        self.values[instance.id]["_modified"] = modified
        if value_context:
            self.values[instance.id]["_context"] = value_context
        self.values[instance.id] = pickle_deepcopy(self.values[instance.id])
        if self.autocommit:
            self.commit_changed_to_source()

fetch_from_source(instances, include_newer=False) #

Fetch cached data from source cache.

Fetch cached values that match modification timestamp of corresponding instance.

Source code in src/apps/cache/serializer_cache.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def fetch_from_source(self, instances: List[models.Model], include_newer=False):
    """Fetch cached data from source cache.

    Fetch cached values that match modification timestamp of corresponding instance.
    """
    cached_values = self.get_source_cache().get_many([instance.id for instance in instances])
    for instance in instances:
        if cached := cached_values.get(instance.id):
            modified = getattr(instance, self.modified_attr)
            if cached["_modified"] == modified:
                self.values[instance.id] = cached
            if include_newer and cached["_modified"] > modified:
                # Include cache entries that are newer than instance
                self.values[instance.id] = cached

get_source_cache() classmethod #

Source cache is determined dynamically so it can be changed in tests.

Source code in src/apps/cache/serializer_cache.py 
40
41
42
43
@classmethod
def get_source_cache(cls) -> BaseCache:
    """Source cache is determined dynamically so it can be changed in tests."""
    return caches[cls.cache_name]

set_value(instance, value, value_context=None, only_if_modified=False) #

Set per-instance value to cache.

The in-memory value is not copied so care needs to be taken to avoid modifying it after using set_value.

Source code in src/apps/cache/serializer_cache.py 
 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
def set_value(
    self,
    instance: models.Model,
    value: dict,
    value_context: Optional[dict] = None,
    only_if_modified=False,
):
    """Set per-instance value to cache.

    The in-memory value is not copied so care needs to be taken
    to avoid modifying it after using set_value.
    """
    modified = getattr(instance, self.modified_attr)

    if only_if_modified:
        entry = self.values.get(instance.id) or {}
        if entry_modified := entry.get("_modified"):
            if modified <= entry_modified:
                return  # Instance is same or older than existing cached entry

    self.changed.add(instance.id)
    self.values[instance.id] = {k: v for k, v in value.items() if k in self.cached_fields}
    self.values[instance.id]["_modified"] = modified
    if value_context:
        self.values[instance.id]["_context"] = value_context
    self.values[instance.id] = pickle_deepcopy(self.values[instance.id])
    if self.autocommit:
        self.commit_changed_to_source()

SerializerCacheSerializer#

Bases: Serializer

Serializer with support for partially cached field values.

Only fields in cached_fields are cached, so cache invalidation is needed only when a value in cached_fields changes.

Usage example: 

class SerializerCache:
    cache_name = "default" # Use default Django cache
    cached_fields = {"field1", "field2"}  # Which fields are cached
    modified_attr = "modified" # Model field containing timestamp of latest modification

class SomeSerializer(ModelSerializer, SerializerCacheSerializer):
    ...

    def to_representation(self, instance):
        ret = super().to_representation(instance)
        if cache := get_serializer_cache():
            # Save serialized value to serializer cache
            cache.set_value(instance, ret, only_if_modified=True)
        return ret

# Serialize instances using cached values
instances = SomeModel.objects.filter(...)
cache = SerializerCache(instances)
serializer = SomeSerializer(instances, cache=cache, many=True)
serialized_values = serializer.data

Source code in src/apps/cache/serializer_cache.py 
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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
class SerializerCacheSerializer(serializers.Serializer):
    """Serializer with support for partially cached field values.

    Only fields in cached_fields are cached, so cache invalidation
    is needed only when a value in cached_fields changes.

    Usage example:
    ```
    class SerializerCache:
        cache_name = "default" # Use default Django cache
        cached_fields = {"field1", "field2"}  # Which fields are cached
        modified_attr = "modified" # Model field containing timestamp of latest modification

    class SomeSerializer(ModelSerializer, SerializerCacheSerializer):
        ...

        def to_representation(self, instance):
            ret = super().to_representation(instance)
            if cache := get_serializer_cache():
                # Save serialized value to serializer cache
                cache.set_value(instance, ret, only_if_modified=True)
            return ret

    # Serialize instances using cached values
    instances = SomeModel.objects.filter(...)
    cache = SerializerCache(instances)
    serializer = SomeSerializer(instances, cache=cache, many=True)
    serialized_values = serializer.data
    ```
    """

    def __init__(self, *args, cache: Optional[SerializerCacheBase] = None, **kwargs):
        self.cache = cache
        super().__init__(*args, **kwargs)

    def get_cached_field_sources(self) -> list:
        """List sources of cached fields."""
        if not self.cache:
            return []

        field_names = self.cache.cached_fields
        model_fields = [
            self.fields[name].source or name for name in field_names if name in self.fields
        ]
        return model_fields

    def to_representation(self, instance) -> dict:
        """Serialization modified to support cached values."""
        ret = {}
        fields = self._readable_fields

        cached_values = None
        cached_fields = None
        if cache := self.cache:
            cached_values = cache.get_value(instance)
            cached_fields = cache.cached_fields

        for field in fields:
            # Use cached value for field if available
            field_name = field.field_name
            if cached_values is not None and field_name in cached_fields:
                if field_name in cached_values:
                    ret[field_name] = cached_values[field_name]
                continue

            # Rest of the method works like normal Serializer.to_representation
            try:
                attribute = field.get_attribute(instance)
            except serializers.SkipField:
                continue

            check_for_none = (
                attribute.pk if isinstance(attribute, serializers.PKOnlyObject) else attribute
            )
            if check_for_none is None:
                ret[field_name] = None
            else:
                ret[field_name] = field.to_representation(attribute)

        return ret

get_cached_field_sources() #

List sources of cached fields.

Source code in src/apps/cache/serializer_cache.py 
152
153
154
155
156
157
158
159
160
161
def get_cached_field_sources(self) -> list:
    """List sources of cached fields."""
    if not self.cache:
        return []

    field_names = self.cache.cached_fields
    model_fields = [
        self.fields[name].source or name for name in field_names if name in self.fields
    ]
    return model_fields

to_representation(instance) #

Serialization modified to support cached values.

Source code in src/apps/cache/serializer_cache.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def to_representation(self, instance) -> dict:
    """Serialization modified to support cached values."""
    ret = {}
    fields = self._readable_fields

    cached_values = None
    cached_fields = None
    if cache := self.cache:
        cached_values = cache.get_value(instance)
        cached_fields = cache.cached_fields

    for field in fields:
        # Use cached value for field if available
        field_name = field.field_name
        if cached_values is not None and field_name in cached_fields:
            if field_name in cached_values:
                ret[field_name] = cached_values[field_name]
            continue

        # Rest of the method works like normal Serializer.to_representation
        try:
            attribute = field.get_attribute(instance)
        except serializers.SkipField:
            continue

        check_for_none = (
            attribute.pk if isinstance(attribute, serializers.PKOnlyObject) else attribute
        )
        if check_for_none is None:
            ret[field_name] = None
        else:
            ret[field_name] = field.to_representation(attribute)

    return ret