Cache

A simple cache using the database.

The Plain Cache stores JSON-serializable values in a CachedItem model. Cached data can be set to expire after a certain amount of time.

Access to the cache is provided through the Cached class.

from plain.cache import Cached


cached = Cached("my-cache-key")

if cached.exists():
    print("Cache hit and not expired!")
    print(cached.value)
else:
    print("Cache miss!")
    cached.set("a JSON-serializable value", expiration=60)

# Delete the item if you need to
cached.delete()

Expired cache items can be cleared with plain cache clear-expired. You can run this on a schedule through various cron-like tools or plain-worker.

Installation

Add plain.cache to your INSTALLED_PACKAGES:

# app/settings.py
INSTALLED_PACKAGES = [
    # ...
    "plain.cache",
]

CLI

plain cache clear-expired - Clear all expired cache items
plain cache clear-all - Clear all cache items
plain cache stats - Show cache statistics

Source

plain-cache

plain

cache

core.py

 1from datetime import datetime, timedelta
 2from functools import cached_property
 3
 4from plain.models import IntegrityError
 5from plain.utils import timezone
 6
 7
 8class Cached:
 9    """Store and retrieve cached items."""
10
11    def __init__(self, key: str) -> None:
12        self.key = key
13
14        # So we can import Cached in __init__.py
15        # without getting the packages not ready error...
16        from .models import CachedItem
17
18        self._model_class = CachedItem
19
20    @cached_property
21    def _model_instance(self):
22        try:
23            return self._model_class.objects.get(key=self.key)
24        except self._model_class.DoesNotExist:
25            return None
26
27    def reload(self) -> None:
28        if hasattr(self, "_model_instance"):
29            del self._model_instance
30
31    def _is_expired(self):
32        if not self._model_instance:
33            return True
34
35        if not self._model_instance.expires_at:
36            return False
37
38        return self._model_instance.expires_at < timezone.now()
39
40    def exists(self) -> bool:
41        if self._model_instance is None:
42            return False
43
44        return not self._is_expired()
45
46    @property
47    def value(self):
48        if not self.exists():
49            return None
50
51        return self._model_instance.value
52
53    def set(self, value, expiration: datetime | timedelta | int | float | None = None):
54        defaults = {
55            "value": value,
56        }
57
58        if isinstance(expiration, int | float):
59            defaults["expires_at"] = timezone.now() + timedelta(seconds=expiration)
60        elif isinstance(expiration, timedelta):
61            defaults["expires_at"] = timezone.now() + expiration
62        elif isinstance(expiration, datetime):
63            defaults["expires_at"] = expiration
64        else:
65            # Keep existing expires_at value or None
66            pass
67
68        # Make sure expires_at is timezone aware
69        if defaults["expires_at"] and not timezone.is_aware(defaults["expires_at"]):
70            defaults["expires_at"] = timezone.make_aware(defaults["expires_at"])
71
72        try:
73            item, _ = self._model_class.objects.update_or_create(
74                key=self.key, defaults=defaults
75            )
76        except IntegrityError:
77            # Most likely a race condition in creating the item,
78            # so trying again should do an update
79            item, _ = self._model_class.objects.update_or_create(
80                key=self.key, defaults=defaults
81            )
82
83        self.reload()
84        return item.value
85
86    def delete(self) -> bool:
87        if not self._model_instance:
88            # A no-op, but a return value you can use to know whether it did anything
89            return False
90
91        self._model_instance.delete()
92        self.reload()
93        return True