Start new CachedItem model

Author: davegaeddert

bolt-cache/bolt/cache/cli.py

@@ -0,0 +1,40 @@
+import click
+
+from .models import CachedItem
+
+
+@click.group()
+def cli():
+    pass
+
+
+@cli.command()
+def clear_expired():
+    click.echo("Clearing expired cache items...")
+    result = CachedItem.objects.expired().delete()
+    click.echo(f"Deleted {result[0]} expired cache items.")
+
+
+@cli.command()
+@click.option("--force", is_flag=True)
+def clear_all(force):
+    if not force and not click.confirm(
+        "Are you sure you want to delete all cache items?"
+    ):
+        return
+    click.echo("Clearing all cache items...")
+    result = CachedItem.objects.all().delete()
+    click.echo(f"Deleted {result[0]} cache items.")
+
+
+@cli.command()
+def stats():
+    total = CachedItem.objects.count()
+    expired = CachedItem.objects.expired().count()
+    unexpired = CachedItem.objects.unexpired().count()
+    forever = CachedItem.objects.forever().count()
+
+    click.echo(f"Total: {click.style(total, bold=True)}")
+    click.echo(f"Expired: {click.style(expired, bold=True)}")
+    click.echo(f"Unexpired: {click.style(unexpired, bold=True)}")
+    click.echo(f"Forever: {click.style(forever, bold=True)}")

bolt-cache/bolt/cache/config.py

@@ -0,0 +1,7 @@
+from bolt.packages import PackageConfig
+
+
+class BoltCacheConfig(PackageConfig):
+    default_auto_field = "bolt.db.models.BigAutoField"
+    name = "bolt.cache"
+    label = "boltcache"

bolt-cache/bolt/cache/core.py

@@ -0,0 +1,74 @@
+from datetime import datetime, timedelta
+from functools import cached_property
+
+from bolt.utils import timezone
+
+from .models import CachedItem
+
+
+class Cached:
+    def __init__(self, key):
+        self.key = key
+
+    @cached_property
+    def _model_instance(self):
+        try:
+            return CachedItem.objects.get(key=self.key)
+        except CachedItem.DoesNotExist:
+            return None
+
+    def reload(self):
+        if hasattr(self, "_model_instance"):
+            del self._model_instance
+
+    def _is_expired(self):
+        if not self._model_instance:
+            return True
+
+        if not self._model_instance.expires_at:
+            return False
+
+        return self._model_instance.expires_at < timezone.now()
+
+    def exists(self):
+        if self._model_instance is None:
+            return False
+
+        return not self._is_expired()
+
+    @property
+    def value(self):
+        if not self.exists():
+            return None
+
+        return self._model_instance.value
+
+    def set(self, value, expiration: datetime | timedelta | int | None = None):
+        defaults = {
+            "value": value,
+        }
+
+        if isinstance(expiration, int):
+            defaults["expires_at"] = timezone.now() + timedelta(seconds=expiration)
+        elif isinstance(expiration, timedelta):
+            defaults["expires_at"] = timezone.now() + expiration
+        elif isinstance(expiration, datetime):
+            defaults["expires_at"] = expiration
+        else:
+            # Keep existing expires_at value or None
+            pass
+
+        item, _ = CachedItem.objects.update_or_create(key=self.key, defaults=defaults)
+
+        self.reload()
+
+        return item.value
+
+    def delete(self):
+        if not self._model_instance:
+            # A no-op, but a return value you can use to know whether it did anything
+            return False
+
+        self._model_instance.delete()
+        self.reload()
+        return True

bolt-cache/bolt/cache/migrations/0001_initial.py

@@ -0,0 +1,34 @@
+# Generated by Bolt 5.0.dev20231127233940 on 2023-12-22 03:47
+
+from bolt.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    initial = True
+
+    dependencies = []
+
+    operations = [
+        migrations.CreateModel(
+            name="CacheItem",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("key", models.CharField(max_length=255, unique=True)),
+                ("value", models.JSONField(blank=True, null=True)),
+                (
+                    "expires_at",
+                    models.DateTimeField(blank=True, db_index=True, null=True),
+                ),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+            ],
+        ),
+    ]

bolt-cache/bolt/cache/migrations/0002_rename_cacheitem_cacheditem.py

@@ -0,0 +1,16 @@
+# Generated by Bolt 5.0.dev20231127233940 on 2023-12-22 17:40
+
+from bolt.db import migrations
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("boltcache", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.RenameModel(
+            old_name="CacheItem",
+            new_name="CachedItem",
+        ),
+    ]

bolt-cache/bolt/cache/migrations/__init__.py

@@ -0,0 +1,26 @@
+from bolt.db import models
+from bolt.utils import timezone
+
+
+class CachedItemQuerySet(models.QuerySet):
+    def expired(self):
+        return self.filter(expires_at__lt=timezone.now())
+
+    def unexpired(self):
+        return self.filter(expires_at__gte=timezone.now())
+
+    def forever(self):
+        return self.filter(expires_at=None)
+
+
+class CachedItem(models.Model):
+    key = models.CharField(max_length=255, unique=True)
+    value = models.JSONField(blank=True, null=True)
+    expires_at = models.DateTimeField(blank=True, null=True, db_index=True)
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+
+    objects = CachedItemQuerySet.as_manager()
+
+    def __str__(self) -> str:
+        return self.key