Performance: scope treenode's cache keys per model

CHANGELOG.md

@@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+-   Scope treenode's internal cache keys per model instead of sharing
+    them across every treenode model in the app.
+
 ## [0.25.0](https://github.com/fabiocaccamo/django-treenode/releases/tag/0.25.0) - 2026-08-26
 -   Add `Django 6.1` support.
 -   Drop `Django < 4.2` support. #220

tests/test_cache_isolation.py

@@ -0,0 +1,43 @@
+from django.test import TestCase
+
+from tests.models import Category, CategoryFixtures
+from treenode.cache import _get_cache, clear_cache, query_cache
+from treenode.signals import no_signals
+
+
+class TreeNodeCacheIsolationTestCase(TestCase):
+    def setUp(self):
+        Category.delete_tree()
+        CategoryFixtures.delete_tree()
+
+    def tearDown(self):
+        Category.delete_tree()
+        CategoryFixtures.delete_tree()
+
+    def test_clearing_one_models_cache_does_not_touch_another_models(self):
+        with no_signals():
+            Category.objects.create(name="cat-a")
+            CategoryFixtures.objects.create(name="fixture-a")
+        Category.update_tree()
+        CategoryFixtures.update_tree()
+
+        # populate both models' caches
+        category_cached = query_cache(Category)
+        fixtures_cached = query_cache(CategoryFixtures)
+        self.assertEqual(len(category_cached), 1)
+        self.assertEqual(len(fixtures_cached), 1)
+
+        # clearing Category's cache must not touch CategoryFixtures' entry
+        clear_cache(Category)
+        c = _get_cache()
+        self.assertIsNone(c.get(f"treenode_list:{Category._meta.label_lower}"))
+        self.assertIsNone(c.get(f"treenode_dict:{Category._meta.label_lower}"))
+        self.assertIsNotNone(
+            c.get(f"treenode_list:{CategoryFixtures._meta.label_lower}")
+        )
+        self.assertIsNotNone(
+            c.get(f"treenode_dict:{CategoryFixtures._meta.label_lower}")
+        )
+
+        # and CategoryFixtures' cached data is still correct
+        self.assertEqual(query_cache(CategoryFixtures), fixtures_cached)

treenode/cache.py

@@ -1,5 +1,4 @@
 import logging
-from collections import defaultdict
 
 from django.conf import settings
 from django.core.cache import cache as default_cache
@@ -19,57 +18,59 @@ def _get_cache_name():
     return "treenode" if "treenode" in settings.CACHES else "default"
 
 
-def _get_cached_collection(key, dict_cls):
+def _cache_key(cls, suffix):
+    return f"treenode_{suffix}:{cls._meta.label_lower}"
+
+
+def _get_cached_collection(cls, suffix, empty_factory):
     c = _get_cache()
+    key = _cache_key(cls, suffix)
     value = c.get(key, None)
     if value is None:
-        value = defaultdict(dict_cls)
+        value = empty_factory()
         c.set(key, value)
     return value
 
 
-def _get_cached_collections():
-    ls = _get_cached_collection("treenode_list", list)
-    d = _get_cached_collection("treenode_dict", dict)
+def _get_cached_collections(cls):
+    ls = _get_cached_collection(cls, "list", list)
+    d = _get_cached_collection(cls, "dict", dict)
     return (ls, d)
 
 
-def _set_cached_collections(ls, d):
+def _set_cached_collections(cls, ls, d):
     c = _get_cache()
-    c.set("treenode_list", ls)
-    c.set("treenode_dict", d)
+    c.set(_cache_key(cls, "list"), ls)
+    c.set(_cache_key(cls, "dict"), d)
 
 
 def clear_cache(cls):
-    ls, d = _get_cached_collections()
-    del ls[cls][:]
-    d[cls].clear()
-    _set_cached_collections(ls, d)
+    c = _get_cache()
+    c.delete(_cache_key(cls, "list"))
+    c.delete(_cache_key(cls, "dict"))
 
 
 def query_cache(cls, pk=None, pks=None):
-    ls, d = _get_cached_collections()
-    if not ls[cls] or not d[cls]:
+    ls, d = _get_cached_collections(cls)
+    if not ls or not d:
         update_cache(cls)
-        ls, d = _get_cached_collections()
+        ls, d = _get_cached_collections(cls)
     if pk is not None:
-        return d[cls].get(str(pk))
+        return d.get(str(pk))
     elif pks is not None:
-        return [d[cls].get(str(pk)) for pk in split_pks(pks)]
+        return [d.get(str(pk)) for pk in split_pks(pks)]
     else:
-        return list(ls[cls])
+        return list(ls)
 
 
 def update_cache(cls):
     objs = list(cls.objects.all())
-    ls, d = _get_cached_collections()
-    ls[cls] = objs
-    d[cls] = {str(obj.pk): obj for obj in objs}
-    _set_cached_collections(ls, d)
+    d = {str(obj.pk): obj for obj in objs}
+    _set_cached_collections(cls, objs, d)
     # ensure cache has been updated correctly
     if len(objs):
-        ls, d = _get_cached_collections()
-        if not ls[cls] or not d[cls]:
+        ls2, d2 = _get_cached_collections(cls)
+        if not ls2 or not d2:
             cn = _get_cache_name()
             msg = (
                 f"Unable to update cache '{cn}', "