perf(plugin): bulk TaggedItem writes + suppress m2m_changed re-fire

NetBox's instance.tags.set([...]) triggers m2m_changed per tagged
instance. The signal handler then (a) writes another ObjectChange row
and (b) re-runs serialize_for_event(instance) for the events_queue.
For changesets with thousands of tagged objects this dominates apply
cost on top of the per-object signal overhead.

Add bulk_tags.apply_tags_bulk(instance_tag_pairs, request) which:
- Bypasses the through-table m2m signal by bulk-writing
  extras_taggeditem rows directly via bulk_create(ignore_conflicts=
  True, batch_size=500). _diode_preload['tag_ids_by_slug'] (from the
  preload step) is reused; any missing slugs are resolved via one
  fallback query.
- Mirrors instance.tags.set([...]) replace-set semantics by deleting
  existing TaggedItem rows for the (content_type, object_id) targets in
  one combined DELETE before the bulk INSERT.
- Re-emits a single enqueue_event(OBJECT_UPDATED) per tagged instance
  so the events_queue payload reflects the post-tag state, without the
  duplicate ObjectChange row.

applier._apply_change pops 'tags' from 'data' before serializer.save()
runs and buffers (instance, tag_input) on request._diode_tag_pairs;
apply_changeset flushes via apply_tags_bulk after its main loop.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: ldrozdz93

netbox_diode_plugin/api/applier.py

@@ -12,6 +12,7 @@
 from extras.models import Tag
 from rest_framework.exceptions import ValidationError as ValidationError
 
+from .bulk_tags import apply_tags_bulk, supports_tags
 from .common import NON_FIELD_ERRORS, Change, ChangeSet, ChangeSetException, ChangeSetResult, ChangeType, error_from_validation_error
 from .matcher import find_existing_object, invalidate_find_obj_entry
 from .plugin_utils import get_object_type_model, legal_fields
@@ -27,6 +28,11 @@ def apply_changeset(change_set: ChangeSet, request) -> ChangeSetResult:
     _validate_change_set(change_set)
     _preload_changeset_cache(change_set, request)
 
+    # Collect (instance, tag_input) pairs as we apply changes; flushed via
+    # apply_tags_bulk after the main loop so we issue one DELETE+bulk INSERT
+    # for the whole changeset instead of per-instance m2m_changed re-fires.
+    request._diode_tag_pairs = []
+
     created = {}
     for change in change_set.changes:
         change_type = change.change_type
@@ -55,6 +61,10 @@ def apply_changeset(change_set: ChangeSet, request) -> ChangeSetResult:
             logger.error(f"Integrity error {object_type}: {e} {data}")
             raise _err(f"created a conflict with an existing {object_type}", object_type, "__all__")
 
+    # Flush deferred tag writes in one bulk pass.
+    apply_tags_bulk(request._diode_tag_pairs, request)
+    request._diode_tag_pairs = []
+
     return ChangeSetResult(
         id=change_set.id,
     )
@@ -163,14 +173,23 @@ def _create_or_find_instance(data: dict, object_type: str, serializer_class, req
 
 
 def _apply_change(data: dict, model_class: models.Model, change: Change, created: dict, request):
+    # Pull tags out of `data` BEFORE serializer.save() so the serializer's
+    # `tag.set([...])` side effect — which fires m2m_changed and triggers a
+    # duplicate ObjectChange + a re-run of serialize_for_event — does not run.
+    # We buffer (instance, tag_input) and flush via apply_tags_bulk after the
+    # main apply_changeset loop.
+    deferred_tags = None
+    if supports_tags(model_class) and isinstance(data.get("tags"), list):
+        deferred_tags = data.pop("tags")
+
     serializer_class = get_serializer_for_model(model_class)
     change_type = change.change_type
+    instance = None
 
     if change_type == ChangeType.CREATE:
         # For component types that may be auto-created from e.g. DeviceType or ModuleType templates,
         # try to find existing object first before attempting to create.
         # This prevents duplicates when components are instantiated during Device/Module save()
-        instance = None
         if _is_auto_created_component(change.object_type):
             instance = _try_find_and_update_existing_instance(data, change.object_type, serializer_class, request)
 
@@ -195,6 +214,10 @@ def _apply_change(data: dict, model_class: models.Model, change: Change, created
             serializer.save()
             invalidate_find_obj_entry(change.object_type, instance.id)
 
+    if deferred_tags is not None and instance is not None:
+        request._diode_tag_pairs.append((instance, deferred_tags))
+
+
 def _set_path(data, path, value):
     path = path.split(".")
     key = path.pop(0)

netbox_diode_plugin/api/bulk_tags.py

@@ -0,0 +1,150 @@
+#!/usr/bin/env python
+# Copyright 2026 NetBox Labs, Inc.
+"""
+Bulk TaggedItem writes for diode apply paths.
+
+PR 4 of BULK-ORM Sprint 1.
+
+NetBox's `instance.tags.set([...])` triggers `m2m_changed` (post_clear +
+post_add) per instance. The signal handler re-fires
+`handle_changed_object`, which (a) writes another ObjectChange UPDATE and
+(b) calls `enqueue_event` again — that re-runs the expensive
+`serialize_for_event(instance)`. Multiplied by every tagged object in a
+changeset this dominates the apply path.
+
+This module bypasses the through-table m2m signal by writing
+`extras_taggeditem` rows directly via `bulk_create` and re-emitting a
+single `enqueue_event` per instance to refresh the queued payload with
+the post-tag state. PR 3's `deferred_changelog` already collapses the
+ObjectChange side; this module collapses the TaggedItem side.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections import defaultdict
+from typing import Any
+
+from core.events import OBJECT_UPDATED
+from django.contrib.contenttypes.models import ContentType
+from django.db.models import Q
+from extras.events import enqueue_event
+from extras.models import Tag, TaggedItem
+from netbox.context import events_queue
+from netbox.models.features import TagsMixin
+
+logger = logging.getLogger(__name__)
+
+
+def supports_tags(model_or_instance) -> bool:
+    """Return True if the model (or instance's class) inherits from TagsMixin."""
+    cls = model_or_instance if isinstance(model_or_instance, type) else type(model_or_instance)
+    return issubclass(cls, TagsMixin)
+
+
+def apply_tags_bulk(instance_tag_pairs: list[tuple[Any, list]], request) -> None:
+    """
+    Write tags for a list of (instance, tag_input) pairs via bulk_create.
+
+    `tag_input` may be a list of slug strings, a list of integer tag IDs, or
+    a list of dicts with a "slug" key (the same shapes the serializer accepts).
+    Tags listed for an instance fully replace its existing tag set, mirroring
+    `instance.tags.set([...])` semantics — but without the m2m_changed re-fire.
+    """
+    if not instance_tag_pairs:
+        return
+
+    preload = getattr(request, "_diode_preload", None) or {}
+    tag_ids_by_slug: dict[str, int] = preload.get("tag_ids_by_slug", {}) or {}
+    _backfill_missing_tag_ids(instance_tag_pairs, tag_ids_by_slug)
+
+    rows, target_keys, instances_to_event = _build_rows(
+        instance_tag_pairs, tag_ids_by_slug,
+    )
+
+    _replace_existing(target_keys)
+    if rows:
+        TaggedItem.objects.bulk_create(rows, ignore_conflicts=True, batch_size=500)
+    _re_emit_events(instances_to_event, request)
+
+
+def _backfill_missing_tag_ids(pairs, tag_ids_by_slug: dict[str, int]) -> None:
+    missing: set[str] = set()
+    for _instance, tag_input in pairs:
+        for raw in tag_input or []:
+            slug = _slug_from(raw)
+            if slug and slug not in tag_ids_by_slug:
+                missing.add(slug)
+    if missing:
+        for tag_id, slug in Tag.objects.filter(slug__in=missing).values_list("id", "slug"):
+            tag_ids_by_slug[slug] = tag_id
+
+
+def _build_rows(pairs, tag_ids_by_slug: dict[str, int]):
+    ct_by_model: dict[type, ContentType] = {}
+    rows: list[TaggedItem] = []
+    target_keys: dict[int, set[int]] = defaultdict(set)
+    instances_to_event: list[Any] = []
+
+    for instance, tag_input in pairs:
+        if instance is None or instance.pk is None:
+            continue
+        ct = _ct_for(instance, ct_by_model)
+        target_keys[ct.id].add(instance.pk)
+        instances_to_event.append(instance)
+        for raw in tag_input or []:
+            tag_id = _resolve_tag_id(raw, tag_ids_by_slug)
+            if tag_id is None:
+                logger.warning("apply_tags_bulk: could not resolve tag %r for %s", raw, instance)
+                continue
+            rows.append(TaggedItem(content_type_id=ct.id, object_id=instance.pk, tag_id=tag_id))
+
+    return rows, target_keys, instances_to_event
+
+
+def _ct_for(instance, ct_by_model: dict[type, ContentType]) -> ContentType:
+    model = type(instance)
+    ct = ct_by_model.get(model)
+    if ct is None:
+        ct = ContentType.objects.get_for_model(model)
+        ct_by_model[model] = ct
+    return ct
+
+
+def _replace_existing(target_keys: dict[int, set[int]]) -> None:
+    if not target_keys:
+        return
+    q = Q()
+    for ct_id, obj_ids in target_keys.items():
+        q |= Q(content_type_id=ct_id, object_id__in=obj_ids)
+    TaggedItem.objects.filter(q).delete()
+
+
+def _re_emit_events(instances, request) -> None:
+    if not instances:
+        return
+    queue = events_queue.get()
+    for instance in instances:
+        try:
+            enqueue_event(queue, instance, request, OBJECT_UPDATED)
+        except Exception as e:
+            logger.warning("apply_tags_bulk: enqueue_event failed for %s: %s", instance, e)
+    events_queue.set(queue)
+
+
+def _slug_from(raw) -> str | None:
+    if isinstance(raw, str):
+        return raw
+    if isinstance(raw, dict):
+        s = raw.get("slug")
+        return s if isinstance(s, str) else None
+    return None
+
+
+def _resolve_tag_id(raw, tag_ids_by_slug: dict[str, int]) -> int | None:
+    if isinstance(raw, int):
+        return raw
+    slug = _slug_from(raw)
+    if slug is not None:
+        return tag_ids_by_slug.get(slug)
+    return None

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

@@ -0,0 +1,133 @@
+#!/usr/bin/env python
+# Copyright 2026 NetBox Labs, Inc.
+"""Diode NetBox Plugin - PR 4 bulk TaggedItem tests."""
+
+import uuid
+from types import SimpleNamespace
+
+from core.models import ObjectChange
+from dcim.models import Site
+from django.contrib.contenttypes.models import ContentType
+from django.db import connection
+from django.test import RequestFactory, TestCase
+from django.test.utils import CaptureQueriesContext
+from extras.models import Tag, TaggedItem
+from netbox.context import current_request, events_queue
+
+from netbox_diode_plugin.api.bulk_tags import apply_tags_bulk
+from netbox_diode_plugin.api.deferred_changelog import deferred_changelog
+from netbox_diode_plugin.plugin_config import get_diode_user
+
+
+def _make_request():
+    rf = RequestFactory()
+    req = rf.post("/x")
+    req.id = uuid.uuid4()
+    req.user = get_diode_user()
+    return req
+
+
+class BulkTaggedItemTestCase(TestCase):
+    """Verifies apply_tags_bulk (PR 4)."""
+
+    def setUp(self):
+        """Create K distinct tags + N untagged sites."""
+        self.tags = [Tag.objects.create(name=f"t{i}", slug=f"t{i}") for i in range(3)]
+        self.sites = [Site.objects.create(name=f"S{i}", slug=f"s{i}") for i in range(4)]
+
+    def test_n_objects_one_bulk_insert(self):
+        """N tagged objects produce 1 INSERT INTO extras_taggeditem (modulo batches)."""
+        req = _make_request()
+        # Mimic preload contract from PR 2.
+        req._diode_preload = {
+            "tag_ids_by_slug": {t.slug: t.id for t in self.tags},
+        }
+        token = current_request.set(req)
+        try:
+            pairs = [(s, [t.slug for t in self.tags]) for s in self.sites]
+            with CaptureQueriesContext(connection) as ctx:
+                apply_tags_bulk(pairs, req)
+            inserts = [
+                q for q in ctx.captured_queries
+                if "INSERT INTO" in q["sql"] and '"extras_taggeditem"' in q["sql"]
+            ]
+            assert len(inserts) == 1, [q["sql"] for q in inserts]
+        finally:
+            current_request.reset(token)
+
+    def test_tag_set_per_object_matches_baseline(self):
+        """Final tag set per instance equals what `instance.tags.set([...])` would produce."""
+        req = _make_request()
+        req._diode_preload = {
+            "tag_ids_by_slug": {t.slug: t.id for t in self.tags},
+        }
+        token = current_request.set(req)
+        try:
+            pairs = [(self.sites[0], ["t0", "t1"]), (self.sites[1], ["t2"])]
+            apply_tags_bulk(pairs, req)
+            ct = ContentType.objects.get_for_model(Site)
+
+            for site, expected_slugs in [(self.sites[0], {"t0", "t1"}), (self.sites[1], {"t2"})]:
+                got = set(
+                    Tag.objects.filter(
+                        id__in=TaggedItem.objects.filter(
+                            content_type=ct, object_id=site.pk
+                        ).values_list("tag_id", flat=True)
+                    ).values_list("slug", flat=True)
+                )
+                assert got == expected_slugs, (site, got, expected_slugs)
+        finally:
+            current_request.reset(token)
+
+    def test_no_extra_audit_row_per_tagged_object(self):
+        """N audit rows for N tagged-only updates — no doubled rows from m2m re-fire."""
+        req = _make_request()
+        req._diode_preload = {
+            "tag_ids_by_slug": {t.slug: t.id for t in self.tags},
+        }
+        token = current_request.set(req)
+        try:
+            ObjectChange.objects.all().delete()
+            with deferred_changelog():
+                # Touch each instance once via a normal save, then bulk-tag.
+                for s in self.sites:
+                    s.description = "touched"
+                    s.save()
+                apply_tags_bulk([(s, ["t0"]) for s in self.sites], req)
+
+            ct = ContentType.objects.get_for_model(Site)
+            row_count = ObjectChange.objects.filter(
+                changed_object_type=ct,
+                changed_object_id__in=[s.pk for s in self.sites],
+            ).count()
+            assert row_count == len(self.sites), row_count
+        finally:
+            current_request.reset(token)
+
+    def test_events_queue_populated_with_post_tag_state(self):
+        """events_queue has one entry per tagged instance, with key matching baseline format."""
+        req = _make_request()
+        req._diode_preload = {
+            "tag_ids_by_slug": {t.slug: t.id for t in self.tags},
+        }
+        token = current_request.set(req)
+        # Reset events queue.
+        q_token = events_queue.set({})
+        try:
+            apply_tags_bulk([(self.sites[0], ["t0"])], req)
+            queued = events_queue.get()
+            expected_key = f"dcim.site:{self.sites[0].pk}"
+            assert expected_key in queued, list(queued.keys())
+        finally:
+            events_queue.reset(q_token)
+            current_request.reset(token)
+
+    def test_empty_pairs_is_noop(self):
+        """No pairs -> no INSERT, no DELETE."""
+        req = _make_request()
+        req._diode_preload = {"tag_ids_by_slug": {}}
+        with CaptureQueriesContext(connection) as ctx:
+            apply_tags_bulk([], req)
+        # An empty input may still touch a tx savepoint, but must not write taggeditem.
+        for q in ctx.captured_queries:
+            assert '"extras_taggeditem"' not in q["sql"], q["sql"]