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