1from __future__ import annotations
 2
 3from typing import Self
 4
 5from plain import models
 6from plain.utils import timezone
 7
 8
 9class CachedItemQuerySet(models.QuerySet["CachedItem"]):
10    def expired(self) -> Self:
11        return self.filter(expires_at__lt=timezone.now())
12
13    def unexpired(self) -> Self:
14        return self.filter(expires_at__gte=timezone.now())
15
16    def forever(self) -> Self:
17        return self.filter(expires_at=None)
18
19
20@models.register_model
21class CachedItem(models.Model):
22    key = models.CharField(max_length=255)
23    value = models.JSONField(required=False, allow_null=True)
24    expires_at = models.DateTimeField(required=False, allow_null=True)
25    created_at = models.DateTimeField(auto_now_add=True)
26    updated_at = models.DateTimeField(auto_now=True)
27
28    query = CachedItemQuerySet()
29
30    model_options = models.Options(
31        indexes=[
32            models.Index(fields=["expires_at"]),
33        ],
34        constraints=[
35            models.UniqueConstraint(
36                fields=["key"], name="plaincache_cacheditem_unique_key"
37            ),
38        ],
39    )
40
41    def __str__(self) -> str:
42        return self.key