perf(plugin): preload ContentType + tag IDs once per changeset

PR 2 of BULK-ORM Sprint 1.

Adds `_preload_changeset_cache(change_set, request)` called at the top of
`apply_changeset`. It walks the changeset once, collects every distinct
object_type and every tag slug, then:

- Calls `ContentType.objects.get_for_models(*models)` — a single query
  that warms Django's per-process content-type cache so subsequent
  per-change `get_for_model()` calls are free.
- Issues one `Tag.objects.filter(slug__in=...).values_list('id','slug')`
  query and stashes `tag_ids_by_slug` on `request._diode_preload`.

PR 4 will reuse `tag_ids_by_slug` to bulk-write `TaggedItem` rows.

Test `test_preload_cache.py` asserts:
- After preload, `get_for_model(Site)` runs in 0 queries.
- Tag IDs for slugs referenced by non-NOOP changes are resolved in <= 2
  queries (1 ContentType warm + 1 Tag fetch).

Author: ldrozdz93

netbox_diode_plugin/api/applier.py

@@ -5,9 +5,11 @@
 
 import logging
 
+from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import ObjectDoesNotExist
 from django.db import models, transaction
 from django.db.utils import IntegrityError
+from extras.models import Tag
 from rest_framework.exceptions import ValidationError as ValidationError
 
 from .common import NON_FIELD_ERRORS, Change, ChangeSet, ChangeSetException, ChangeSetResult, ChangeType, error_from_validation_error
@@ -23,6 +25,7 @@
 def apply_changeset(change_set: ChangeSet, request) -> ChangeSetResult:
     """Apply a change set."""
     _validate_change_set(change_set)
+    _preload_changeset_cache(change_set, request)
 
     created = {}
     for change in change_set.changes:
@@ -56,6 +59,63 @@ def apply_changeset(change_set: ChangeSet, request) -> ChangeSetResult:
         id=change_set.id,
     )
 
+def _preload_changeset_cache(change_set: ChangeSet, request) -> dict:
+    """
+    Warm Django's ContentType cache and prefetch tag IDs once per changeset.
+
+    Without this, every per-change `ContentType.objects.get_for_model(...)` and
+    every `Tag.objects.get(slug=...)` issues its own SQL. Stashing the result on
+    `request._diode_preload` lets later code paths (PR 4 bulk-tag write) reuse
+    it without re-querying.
+    """
+    models_to_warm: dict[str, models.Model] = {}
+    tag_slugs: set[str] = set()
+
+    for change in change_set.changes:
+        if change.change_type == ChangeType.NOOP:
+            continue
+        ot = change.object_type
+        if ot and ot not in models_to_warm:
+            try:
+                models_to_warm[ot] = get_object_type_model(ot)
+            except Exception:
+                # Unknown model — let the main apply path raise the proper error.
+                continue
+        for slug in _iter_tag_slugs(change):
+            tag_slugs.add(slug)
+
+    if models_to_warm:
+        # Populates Django's per-process ContentType cache in a single query.
+        ContentType.objects.get_for_models(*models_to_warm.values())
+
+    tag_ids_by_slug: dict[str, int] = {}
+    if tag_slugs:
+        for tag_id, slug in Tag.objects.filter(slug__in=tag_slugs).values_list("id", "slug"):
+            tag_ids_by_slug[slug] = tag_id
+
+    preload = {
+        "tag_ids_by_slug": tag_ids_by_slug,
+        "models_by_object_type": models_to_warm,
+    }
+    if request is not None:
+        request._diode_preload = preload
+    return preload
+
+
+def _iter_tag_slugs(change: Change):
+    """Yield string tag slugs from a change.data['tags'] list."""
+    if not change.data:
+        return
+    tags = change.data.get("tags")
+    if not isinstance(tags, list):
+        return
+    for t in tags:
+        if isinstance(t, str):
+            yield t
+        elif isinstance(t, dict) and isinstance(t.get("slug"), str):
+            yield t["slug"]
+
+
 def _is_auto_created_component(object_type: str) -> bool:
     """Check if the object type is auto-created from templates."""
     auto_created_components = [

netbox_diode_plugin/tests/v4.5.x/tests/test_preload_cache.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python
+# Copyright 2026 NetBox Labs, Inc.
+"""Diode NetBox Plugin - PR 2 preload cache tests."""
+
+from types import SimpleNamespace
+
+from dcim.models import Site
+from django.contrib.contenttypes.models import ContentType
+from django.db import connection
+from django.test import TestCase
+from django.test.utils import CaptureQueriesContext
+from extras.models import Tag
+
+from netbox_diode_plugin.api.applier import _preload_changeset_cache
+from netbox_diode_plugin.api.common import Change, ChangeSet, ChangeType
+
+
+class PreloadChangesetCacheTestCase(TestCase):
+    """Verifies _preload_changeset_cache (PR 2)."""
+
+    def setUp(self):
+        """Create two tags + a 3-change changeset (one NOOP) for the assertions below."""
+        self.tag_a = Tag.objects.create(name="alpha", slug="alpha")
+        self.tag_b = Tag.objects.create(name="beta", slug="beta")
+
+        self.change_set = ChangeSet(
+            id="11111111-1111-1111-1111-111111111111",
+            changes=[
+                Change(
+                    change_type=ChangeType.CREATE.value,
+                    object_type="dcim.site",
+                    ref_id="r1",
+                    data={"name": "S1", "slug": "s1", "tags": ["alpha", "beta"]},
+                ),
+                Change(
+                    change_type=ChangeType.CREATE.value,
+                    object_type="dcim.site",
+                    ref_id="r2",
+                    # repeated tag — must dedup
+                    data={"name": "S2", "slug": "s2", "tags": ["alpha"]},
+                ),
+                Change(
+                    change_type=ChangeType.NOOP.value,
+                    object_type="dcim.site",
+                    ref_id="r3",
+                    data={"tags": ["ignored"]},
+                ),
+            ],
+        )
+
+    def test_contenttype_cache_hit_after_preload(self):
+        """get_for_model on a preloaded model issues 0 queries."""
+        ContentType.objects.clear_cache()
+        request = SimpleNamespace()
+        _preload_changeset_cache(self.change_set, request)
+
+        with CaptureQueriesContext(connection) as ctx:
+            ContentType.objects.get_for_model(Site)
+        assert len(ctx.captured_queries) == 0, ctx.captured_queries
+
+    def test_tag_ids_collected_in_at_most_two_queries(self):
+        """Preload should resolve tag slugs in <= 2 queries total (1 ContentType + 1 Tag)."""
+        ContentType.objects.clear_cache()
+        request = SimpleNamespace()
+        with CaptureQueriesContext(connection) as ctx:
+            preload = _preload_changeset_cache(self.change_set, request)
+        # NOOP changes are skipped, so "ignored" tag should NOT be loaded.
+        assert preload["tag_ids_by_slug"] == {
+            "alpha": self.tag_a.id,
+            "beta": self.tag_b.id,
+        }
+        # 1 query for ContentType warm-up + 1 query for Tag.filter(...). Allow
+        # a small slack in case Django prefers to issue a savepoint.
+        assert len(ctx.captured_queries) <= 2, ctx.captured_queries
+        assert request._diode_preload is preload