1from __future__ import annotations
 2
 3from datetime import datetime
 4from typing import Any, Self
 5
 6from plain import postgres
 7from plain.postgres import types
 8from plain.utils import timezone
 9
10__all__ = ["CachedItem", "CachedItemQuerySet"]
11
12
13class CachedItemQuerySet(postgres.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@postgres.register_model
25class CachedItem(postgres.Model):
26    key: str = types.TextField(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 = postgres.Options(
35        indexes=[
36            postgres.Index(
37                name="plaincache_cacheditem_expires_at_idx", fields=["expires_at"]
38            ),
39        ],
40        constraints=[
41            postgres.UniqueConstraint(
42                fields=["key"], name="plaincache_cacheditem_unique_key"
43            ),
44        ],
45    )
46
47    def __str__(self) -> str:
48        return self.key