1from datetime import datetime
 2from typing import ClassVar, Self
 3
 4from plain.postgres import Field, types
 5from plain.runtime import settings
 6from plain.utils import timezone
 7
 8from plain import postgres
 9
10__all__ = ["CachedItem", "CachedItemQuerySet"]
11
12
13class CachedItemQuerySet(postgres.QuerySet["CachedItem"]):
14    def live(self) -> Self:
15        """Rows readable right now: never-expiring *or* not-yet-expired.
16
17        This is the filter cache reads use -- an entry past its `expires_at`
18        reads as absent. (Contrast `unexpired()`, which matches only rows with a
19        *future* expiry.)
20        """
21        return self.where(
22            CachedItem.expires_at.is_null() | CachedItem.expires_at.gte(timezone.now())
23        )
24
25    def expired(self) -> Self:
26        return self.where(CachedItem.expires_at.lt(timezone.now()))
27
28    def unexpired(self) -> Self:
29        return self.where(CachedItem.expires_at.gte(timezone.now()))
30
31    def forever(self) -> Self:
32        return self.where(CachedItem.expires_at.is_null())
33
34
35@postgres.register_model
36class CachedItem(postgres.Model):
37    key: Field[str] = types.TextField(max_length=255)
38    # `object`, not `Any`: the cache stores any JSON-serializable value, and
39    # `Field[Any]` would make `Any` satisfy the model-valued `__get__` overload,
40    # so `CachedItem.value` would type as `type[Any]` and lose the field surface
41    # entirely.
42    value: Field[object] = types.JSONField(
43        required=False, allow_null=True, default=None
44    )
45    expires_at: Field[datetime | None] = types.DateTimeField(
46        required=False, allow_null=True, default=None
47    )
48    created_at: Field[datetime] = types.DateTimeField(create_now=True)
49    updated_at: Field[datetime] = types.DateTimeField(create_now=True, update_now=True)
50
51    # ClassVar override so @typed doesn't treat it as a constructor field;
52    # base Model.query is also a ClassVar, so this override is clean.
53    query: ClassVar[CachedItemQuerySet] = CachedItemQuerySet()
54
55    model_options = postgres.Options(
56        indexes=[
57            postgres.Index(
58                name="plaincache_cacheditem_expires_at_idx", fields=["expires_at"]
59            ),
60        ],
61        constraints=[
62            postgres.UniqueConstraint(
63                fields=["key"], name="plaincache_cacheditem_unique_key"
64            ),
65        ],
66        storage_parameters={
67            "autovacuum_vacuum_scale_factor": settings.CACHE_AUTOVACUUM_SCALE_FACTOR,
68            "toast.autovacuum_vacuum_scale_factor": settings.CACHE_TOAST_AUTOVACUUM_SCALE_FACTOR,
69        },
70    )
71
72    def __str__(self) -> str:
73        return self.key