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