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