Fixes #2073 -- Added DatabaseStore for persistent debug data storage. (#2121)

* feat: add DatabaseStore for persistent debug data storage

- Introduced `DatabaseStore` to store debug toolbar data in the database.
- Added `DebugToolbarEntry` model and migrations for persistent storage.
- Updated documentation to include configuration for `DatabaseStore`.
- Added tests for `DatabaseStore` functionality, including CRUD
  operations and cache size enforcement.

Fixes #2073

* refactor: rename DebugToolbarEntry to HistoryEntry and more

- Updated model name from `DebugToolbarEntry` to `HistoryEntry` to make
  string representations of the app_model less redundant.
- Adjusted verbose names to use translations with `gettext_lazy`.
- Updated all references in `store.py` to use the new model name.
- Modified tests to reflect the model name change.
- Added a test to check the default ordering of the model and make it
  the default ordering in methods reliable.

* Optimize entry creation logic to clean up old entries only when new entries are added

* Wrap creation and update methods in atomic transactions

* Avoid using .set() for database store

This doesn't provide the same utility as it does for the memory
store. We need to use get_or_create to generate the entry in the
database regardless. The middleware will be using .set() to trim
extra requests to avoid overflowing the store removing the need
for save_panel to also do the same.

---------

Co-authored-by: Tim Schilling <schillingt@better-simple.com>

Author: dr-rompecabezas

debug_toolbar/migrations/0001_initial.py

@@ -0,0 +1,24 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    initial = True
+
+    operations = [
+        migrations.CreateModel(
+            name="HistoryEntry",
+            fields=[
+                (
+                    "request_id",
+                    models.UUIDField(primary_key=True, serialize=False),
+                ),
+                ("data", models.JSONField(default=dict)),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+            ],
+            options={
+                "verbose_name": "history entry",
+                "verbose_name_plural": "history entries",
+                "ordering": ["-created_at"],
+            },
+        ),
+    ]

debug_toolbar/migrations/__init__.py

@@ -0,0 +1,16 @@
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+
+
+class HistoryEntry(models.Model):
+    request_id = models.UUIDField(primary_key=True)
+    data = models.JSONField(default=dict)
+    created_at = models.DateTimeField(auto_now_add=True)
+
+    class Meta:
+        verbose_name = _("history entry")
+        verbose_name_plural = _("history entries")
+        ordering = ["-created_at"]
+
+    def __str__(self):
+        return str(self.request_id)

debug_toolbar/models.py

@@ -6,10 +6,12 @@
 from typing import Any
 
 from django.core.serializers.json import DjangoJSONEncoder
+from django.db import transaction
 from django.utils.encoding import force_str
 from django.utils.module_loading import import_string
 
 from debug_toolbar import settings as dt_settings
+from debug_toolbar.models import HistoryEntry
 
 logger = logging.getLogger(__name__)
 
@@ -140,5 +142,86 @@ def panels(cls, request_id: str) -> Any:
             yield panel, deserialize(data)
 
 
+class DatabaseStore(BaseStore):
+    @classmethod
+    def _cleanup_old_entries(cls):
+        """
+        Enforce the cache size limit - keeping only the most recently used entries
+        up to RESULTS_CACHE_SIZE.
+        """
+        # Determine which entries to keep
+        keep_ids = cls.request_ids()
+
+        # Delete all entries not in the keep list
+        if keep_ids:
+            HistoryEntry.objects.exclude(request_id__in=keep_ids).delete()
+
+    @classmethod
+    def request_ids(cls):
+        """Return all stored request ids within the cache size limit"""
+        cache_size = dt_settings.get_config()["RESULTS_CACHE_SIZE"]
+        return list(
+            HistoryEntry.objects.all()[:cache_size].values_list("request_id", flat=True)
+        )
+
+    @classmethod
+    def exists(cls, request_id: str) -> bool:
+        """Check if the given request_id exists in the store"""
+        return HistoryEntry.objects.filter(request_id=request_id).exists()
+
+    @classmethod
+    def set(cls, request_id: str):
+        """Set a request_id in the store and clean up old entries"""
+        with transaction.atomic():
+            # Create the entry if it doesn't exist (ignore otherwise)
+            _, created = HistoryEntry.objects.get_or_create(request_id=request_id)
+
+            # Only enforce cache size limit when new entries are created
+            if created:
+                cls._cleanup_old_entries()
+
+    @classmethod
+    def clear(cls):
+        """Remove all requests from the store"""
+        HistoryEntry.objects.all().delete()
+
+    @classmethod
+    def delete(cls, request_id: str):
+        """Delete the stored request for the given request_id"""
+        HistoryEntry.objects.filter(request_id=request_id).delete()
+
+    @classmethod
+    def save_panel(cls, request_id: str, panel_id: str, data: Any = None):
+        """Save the panel data for the given request_id"""
+        with transaction.atomic():
+            obj, _ = HistoryEntry.objects.get_or_create(request_id=request_id)
+            store_data = obj.data
+            store_data[panel_id] = serialize(data)
+            obj.data = store_data
+            obj.save()
+
+    @classmethod
+    def panel(cls, request_id: str, panel_id: str) -> Any:
+        """Fetch the panel data for the given request_id"""
+        try:
+            data = HistoryEntry.objects.get(request_id=request_id).data
+            panel_data = data.get(panel_id)
+            if panel_data is None:
+                return {}
+            return deserialize(panel_data)
+        except HistoryEntry.DoesNotExist:
+            return {}
+
+    @classmethod
+    def panels(cls, request_id: str) -> Any:
+        """Fetch all panel data for the given request_id"""
+        try:
+            data = HistoryEntry.objects.get(request_id=request_id).data
+            for panel_id, panel_data in data.items():
+                yield panel_id, deserialize(panel_data)
+        except HistoryEntry.DoesNotExist:
+            return {}
+
+
 def get_store() -> BaseStore:
     return import_string(dt_settings.get_config()["TOOLBAR_STORE_CLASS"])

debug_toolbar/store.py

@@ -24,6 +24,8 @@ Serializable (don't include in main)
 * Update all panels to utilize data from ``Panel.get_stats()`` to load content
   to render. Specifically for ``Panel.title`` and ``Panel.nav_title``.
 * Extend example app to contain an async version.
+* Added ``debug_toolbar.store.DatabaseStore`` for persistent debug data
+  storage.
 
 Pending
 -------

docs/changes.rst

@@ -109,7 +109,8 @@ Toolbar options
 
   Default: ``25``
 
-  The toolbar keeps up to this many results in memory.
+  The toolbar keeps up to this many results in memory or persistent storage.
+
 
 .. _ROOT_TAG_EXTRA_ATTRS:
 
@@ -186,6 +187,24 @@ Toolbar options
 
   The path to the class to be used for storing the toolbar's data per request.
 
+  Available store classes:
+
+  * ``debug_toolbar.store.MemoryStore`` - Stores data in memory
+  * ``debug_toolbar.store.DatabaseStore`` - Stores data in the database
+
+  The DatabaseStore provides persistence and automatically cleans up old
+  entries based on the ``RESULTS_CACHE_SIZE`` setting.
+
+  Note: For full functionality, DatabaseStore requires migrations for
+  the debug_toolbar app:
+
+  .. code-block:: bash
+
+      python manage.py migrate debug_toolbar
+
+  For the DatabaseStore to work properly, you need to run migrations for the
+  debug_toolbar app. The migrations create the necessary database table to store
+  toolbar data.
 
 .. _TOOLBAR_LANGUAGE:
 
@@ -394,6 +413,14 @@ Here's what a slightly customized toolbar configuration might look like::
         'SQL_WARNING_THRESHOLD': 100,   # milliseconds
     }
 
+Here's an example of using a persistent store to keep debug data between server
+restarts::
+
+    DEBUG_TOOLBAR_CONFIG = {
+        'TOOLBAR_STORE_CLASS': 'debug_toolbar.store.DatabaseStore',
+        'RESULTS_CACHE_SIZE': 100,  # Store up to 100 requests
+    }
+
 Theming support
 ---------------
 The debug toolbar uses CSS variables to define fonts and colors. This allows

docs/configuration.rst

@@ -0,0 +1,32 @@
+import uuid
+
+from django.test import TestCase
+
+from debug_toolbar.models import HistoryEntry
+
+
+class HistoryEntryTestCase(TestCase):
+    def test_str_method(self):
+        test_uuid = uuid.uuid4()
+        entry = HistoryEntry(request_id=test_uuid)
+        self.assertEqual(str(entry), str(test_uuid))
+
+    def test_data_field_default(self):
+        """Test that the data field defaults to an empty dict"""
+        entry = HistoryEntry(request_id=uuid.uuid4())
+        self.assertEqual(entry.data, {})
+
+    def test_model_persistence(self):
+        """Test saving and retrieving a model instance"""
+        test_uuid = uuid.uuid4()
+        entry = HistoryEntry(request_id=test_uuid, data={"test": True})
+        entry.save()
+
+        # Retrieve from database and verify
+        saved_entry = HistoryEntry.objects.get(request_id=test_uuid)
+        self.assertEqual(saved_entry.data, {"test": True})
+        self.assertEqual(str(saved_entry), str(test_uuid))
+
+    def test_default_ordering(self):
+        """Test that the default ordering is by created_at in descending order"""
+        self.assertEqual(HistoryEntry._meta.ordering, ["-created_at"])

tests/test_models.py

@@ -1,3 +1,5 @@
+import uuid
+
 from django.test import TestCase
 from django.test.utils import override_settings
 
@@ -109,3 +111,121 @@ def test_get_store(self):
     )
     def test_get_store_with_setting(self):
         self.assertIs(store.get_store(), StubStore)
+
+
+class DatabaseStoreTestCase(TestCase):
+    @classmethod
+    def setUpTestData(cls) -> None:
+        cls.store = store.DatabaseStore
+
+    def tearDown(self) -> None:
+        self.store.clear()
+
+    def test_ids(self):
+        id1 = str(uuid.uuid4())
+        id2 = str(uuid.uuid4())
+        self.store.set(id1)
+        self.store.set(id2)
+        # Convert the UUIDs to strings for comparison
+        request_ids = {str(id) for id in self.store.request_ids()}
+        self.assertEqual(request_ids, {id1, id2})
+
+    def test_exists(self):
+        missing_id = str(uuid.uuid4())
+        self.assertFalse(self.store.exists(missing_id))
+        id1 = str(uuid.uuid4())
+        self.store.set(id1)
+        self.assertTrue(self.store.exists(id1))
+
+    def test_set(self):
+        id1 = str(uuid.uuid4())
+        self.store.set(id1)
+        self.assertTrue(self.store.exists(id1))
+
+    def test_set_max_size(self):
+        with self.settings(DEBUG_TOOLBAR_CONFIG={"RESULTS_CACHE_SIZE": 1}):
+            # Clear any existing entries first
+            self.store.clear()
+
+            # Add first entry
+            id1 = str(uuid.uuid4())
+            self.store.set(id1)
+
+            # Verify it exists
+            self.assertTrue(self.store.exists(id1))
+
+            # Add second entry, which should push out the first one due to size limit=1
+            id2 = str(uuid.uuid4())
+            self.store.set(id2)
+
+            # Verify only the bar entry exists now
+            # Convert the UUIDs to strings for comparison
+            request_ids = {str(id) for id in self.store.request_ids()}
+            self.assertEqual(request_ids, {id2})
+            self.assertFalse(self.store.exists(id1))
+
+    def test_clear(self):
+        id1 = str(uuid.uuid4())
+        self.store.save_panel(id1, "bar.panel", {"a": 1})
+        self.store.clear()
+        self.assertEqual(list(self.store.request_ids()), [])
+        self.assertEqual(self.store.panel(id1, "bar.panel"), {})
+
+    def test_delete(self):
+        id1 = str(uuid.uuid4())
+        self.store.save_panel(id1, "bar.panel", {"a": 1})
+        self.store.delete(id1)
+        self.assertEqual(list(self.store.request_ids()), [])
+        self.assertEqual(self.store.panel(id1, "bar.panel"), {})
+        # Make sure it doesn't error
+        self.store.delete(id1)
+
+    def test_save_panel(self):
+        id1 = str(uuid.uuid4())
+        self.store.save_panel(id1, "bar.panel", {"a": 1})
+        self.assertTrue(self.store.exists(id1))
+        self.assertEqual(self.store.panel(id1, "bar.panel"), {"a": 1})
+
+    def test_update_panel(self):
+        id1 = str(uuid.uuid4())
+        self.store.save_panel(id1, "test.panel", {"original": True})
+        self.assertEqual(self.store.panel(id1, "test.panel"), {"original": True})
+
+        # Update the panel
+        self.store.save_panel(id1, "test.panel", {"updated": True})
+        self.assertEqual(self.store.panel(id1, "test.panel"), {"updated": True})
+
+    def test_panels_nonexistent_request(self):
+        missing_id = str(uuid.uuid4())
+        panels = dict(self.store.panels(missing_id))
+        self.assertEqual(panels, {})
+
+    def test_panel(self):
+        id1 = str(uuid.uuid4())
+        missing_id = str(uuid.uuid4())
+        self.assertEqual(self.store.panel(missing_id, "missing"), {})
+        self.store.save_panel(id1, "bar.panel", {"a": 1})
+        self.assertEqual(self.store.panel(id1, "bar.panel"), {"a": 1})
+
+    def test_panels(self):
+        id1 = str(uuid.uuid4())
+        self.store.save_panel(id1, "panel1", {"a": 1})
+        self.store.save_panel(id1, "panel2", {"b": 2})
+        panels = dict(self.store.panels(id1))
+        self.assertEqual(len(panels), 2)
+        self.assertEqual(panels["panel1"], {"a": 1})
+        self.assertEqual(panels["panel2"], {"b": 2})
+
+    def test_cleanup_old_entries(self):
+        # Create multiple entries
+        ids = [str(uuid.uuid4()) for _ in range(5)]
+        for id in ids:
+            self.store.save_panel(id, "test.panel", {"test": True})
+
+        # Set a small cache size
+        with self.settings(DEBUG_TOOLBAR_CONFIG={"RESULTS_CACHE_SIZE": 2}):
+            # Trigger cleanup
+            self.store._cleanup_old_entries()
+
+            # Check that only the most recent 2 entries remain
+            self.assertEqual(len(list(self.store.request_ids())), 2)

tests/test_store.py

@@ -25,7 +25,7 @@ deps =
     pygments
     selenium>=4.8.0
     sqlparse
-    django-csp
+    django-csp<4.0
 passenv=
     CI
     COVERAGE_ARGS