Enhance WarnErrorOptionsV2 to support per-instance targeting with discriminators

.changes/unreleased/Features-20260305-210043.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Enhance WarnErrorOptionsV2 to support per-instance targeting with discriminators
+time: 2026-03-05T21:00:43.174478-08:00
+custom:
+    Author: jairus-m
+    Issue: "358"

dbt_common/events/base_types.py

@@ -113,6 +113,12 @@ def to_json(self) -> str:
     def level_tag(self) -> EventLevel:
         return EventLevel.DEBUG
 
+    def discriminator(self) -> Optional[str]:
+        """Return a per-instance sub-identifier for fine-grained warn_error targeting.
+        Override in subclasses where multiple instances share one event class.
+        """
+        return None
+
     def message(self) -> str:
         raise Exception("message() not implemented for event")
 

dbt_common/events/types.py

@@ -37,6 +37,9 @@
 
 
 class BehaviorChangeEvent(WarnLevel):
+    def discriminator(self) -> str:
+        return self.flag_name
+
     def code(self) -> str:
         return "D000"
 

dbt_common/helper_types.py

@@ -115,9 +115,14 @@ class WarnErrorOptionsV2(dbtClassMixin):
     valid_error_names: a set of event names that can be named in error, warn, and silence.
 
     In a hierarchy of configuration, the following rules apply:
-    1. named > Deprecations > "all"/"*"
+    1. ClassName:<discriminator> > named > Deprecations > "all"/"*"
     2. silence > warn > error
     3. (1) > (2)
+
+    ClassName:<discriminator> entries target individual event instances by their discriminator()
+    return value. They take precedence over class-level targeting, Deprecations, and all/*.
+    The discriminator must be non-empty. Any event class that implements discriminator() can
+    opt into this per-instance targeting.
     """
 
     ERROR_ALL = ("all", "*")
@@ -137,7 +142,7 @@ def __init__(
         self._valid_error_names: Set[str] = valid_error_names or set()
         self._valid_error_names.add(self.DEPRECATIONS)
 
-        # We can't do `= error or []` because if someone passes in an empty list, and latter appends to that list
+        # We can't do `= error or []` because if someone passes in an empty list, and later appends to that list
         # they would expect references to the original list to be updated.
         self.error = error if error is not None else []
         self.warn = warn if warn is not None else []
@@ -150,10 +155,12 @@ def __post_init__(self):
         if isinstance(self.error, str) and self.error not in self.ERROR_ALL:
             raise ValidationError(f"error must be one of {self.ERROR_ALL} or a list of strings")
 
-        # To specify `warn`, one of the following must be true
+        # <event_name>:<flag_name> items are always allowed in `warn` — they are specific
+        # per-instance overrides. Non-discriminated items require a broad `error` setting:
         # 1. `error` must be "all"/"*"
         # 2. "deprecations" must be in either `error` or `silence`.
-        if self.warn and not (
+        non_discriminated_warns = [w for w in self.warn if ":" not in w]
+        if non_discriminated_warns and not (
             self.error in self.ERROR_ALL
             or self.DEPRECATIONS in self.error
             or self.DEPRECATIONS in self.silence
@@ -174,7 +181,15 @@ def __post_init__(self):
 
     def _validate_items(self, items: List[str]):
         for item in items:
-            if item not in self._valid_error_names:
+            if ":" in item:
+                class_name, discriminator = item.split(":", 1)
+                if not discriminator:
+                    raise ValidationError(
+                        f"'{item}' is missing a discriminator value. Use 'ClassName:value'."
+                    )
+                if class_name not in self._valid_error_names:
+                    raise ValidationError(f"'{class_name}' is not a valid dbt error name.")
+            elif item not in self._valid_error_names:
                 raise ValidationError(f"{item} is not a valid dbt error name.")
 
     @property
@@ -207,7 +222,7 @@ def _error_as_deprecation(self, event: Optional[BaseEvent]) -> bool:
         )
 
     def _warn_as_deprecation(self, event: Optional[BaseEvent]) -> bool:
-        """Is the event a deprecation, and if so should it be treated as an warning?"""
+        """Is the event a deprecation, and if so should it be treated as a warning?"""
         return (
             event is not None and event.code().startswith("D") and self.DEPRECATIONS in self.warn
         )
@@ -220,6 +235,17 @@ def _silence_as_deprecation(self, event: Optional[BaseEvent]) -> bool:
             and self.DEPRECATIONS in self.silence
         )
 
+    def _discriminated_match(
+        self, event: Optional[BaseEvent], target_list: Union[str, List[str]]
+    ) -> bool:
+        """Does this event's ClassName:discriminator() appear in target_list?"""
+        if event is None or isinstance(target_list, str):
+            return False
+        disc = event.discriminator()
+        if disc is None:
+            return False
+        return f"{type(event).__name__}:{disc}" in target_list
+
     def errors(self, item_name: Union[str, BaseEvent]) -> bool:
         """Should the event be treated as an error?
 
@@ -242,13 +268,22 @@ def errors(self, item_name: Union[str, BaseEvent]) -> bool:
         deprecation_elsewhere = self._warn_as_deprecation(event) or self._silence_as_deprecation(
             event
         )
+        discriminated_elsewhere = self._discriminated_match(
+            event, self.warn
+        ) or self._discriminated_match(event, self.silence)
 
         # Calculate result
-        if self._named_error(event_name) and not named_elsewhere:
+        if self._discriminated_match(event, self.error) and not discriminated_elsewhere:
+            return True
+        elif self._named_error(event_name) and not (named_elsewhere or discriminated_elsewhere):
             return True
-        elif self._error_as_deprecation(event) and not (named_elsewhere or deprecation_elsewhere):
+        elif self._error_as_deprecation(event) and not (
+            named_elsewhere or deprecation_elsewhere or discriminated_elsewhere
+        ):
             return True
-        elif self._error_all() and not (named_elsewhere or deprecation_elsewhere):
+        elif self._error_all() and not (
+            named_elsewhere or deprecation_elsewhere or discriminated_elsewhere
+        ):
             return True
         else:
             return False
@@ -260,7 +295,8 @@ def includes(self, item_name: Union[str, BaseEvent]) -> bool:
     def silenced(self, item_name: Union[str, BaseEvent]) -> bool:
         """Is the event silenced?
 
-        An event silenced if any of the following are true:
+        An event is silenced if any of the following are true:
+        - The event's ClassName:discriminator() appears in `silence` (silence > warn > error)
         - The event is named in `silence`
         - "Deprecations" is in `silence` and the event is not named in `error` or `warn`
         """
@@ -274,11 +310,18 @@ def silenced(self, item_name: Union[str, BaseEvent]) -> bool:
 
         # Pre-compute checks that will be used multiple times
         named_elsewhere = self._named_error(event_name) or self._named_warn(event_name)
+        discriminated_elsewhere = self._discriminated_match(
+            event, self.error
+        ) or self._discriminated_match(event, self.warn)
 
-        # Calculate result
-        if self._named_silence(event_name):
+        # Calculate result (silence wins over warn/error when same specificity)
+        if self._discriminated_match(event, self.silence):
             return True
-        elif self._silence_as_deprecation(event) and not named_elsewhere:
+        elif self._named_silence(event_name) and not discriminated_elsewhere:
+            return True
+        elif self._silence_as_deprecation(event) and not (
+            named_elsewhere or discriminated_elsewhere
+        ):
             return True
         else:
             return False

tests/unit/test_helper_types.py

@@ -276,3 +276,133 @@ def test_silenced(
     def test_dictification(self) -> None:
         my_options = WarnErrorOptionsV2(error=[], warn=[], silence=[])
         assert my_options.to_dict() == {"error": [], "warn": [], "silence": []}
+
+    def test_discriminated_validation_accepts_valid_entry(self) -> None:
+        WarnErrorOptionsV2(
+            silence=["BehaviorChangeEvent:anything"],
+            valid_error_names={"BehaviorChangeEvent"},
+        )
+        WarnErrorOptionsV2(
+            error=["BehaviorChangeEvent:use_mat_v2"],
+            valid_error_names={"BehaviorChangeEvent"},
+        )
+
+    def test_discriminated_empty_discriminator_raises(self) -> None:
+        with pytest.raises(ValidationError, match="missing a discriminator value"):
+            WarnErrorOptionsV2(silence=["BehaviorChangeEvent:"])
+
+    def test_discriminated_warn_allowed_without_broad_error(self) -> None:
+        # ClassName:<discriminator> in warn should not require error=all/* or Deprecations
+        WarnErrorOptionsV2(
+            error=["BehaviorChangeEvent"],
+            warn=["BehaviorChangeEvent:use_mat_v2"],
+            silence=[],
+            valid_error_names={"BehaviorChangeEvent"},
+        )
+
+    def test_non_discriminated_warn_still_requires_broad_error(self) -> None:
+        with pytest.raises(ValidationError, match="`warn` can only be specified"):
+            WarnErrorOptionsV2(
+                error=["BehaviorChangeEvent"],
+                warn=["SomeOtherEvent"],
+                silence=[],
+                valid_error_names={"BehaviorChangeEvent", "SomeOtherEvent"},
+            )
+
+    # Per-flag BehaviorChange targeting in errors()
+    @pytest.mark.parametrize(
+        "error,warn,silence,flag_name,expected_errors",
+        [
+            # flag-level error fires when not overridden
+            (["BehaviorChangeEvent:use_mat_v2"], [], [], "use_mat_v2", True),
+            # flag-level silence beats flag-level error
+            (
+                ["BehaviorChangeEvent:use_mat_v2"],
+                [],
+                ["BehaviorChangeEvent:use_mat_v2"],
+                "use_mat_v2",
+                False,
+            ),
+            # flag-level silence beats class-level error
+            (["BehaviorChangeEvent"], [], ["BehaviorChangeEvent:use_mat_v2"], "use_mat_v2", False),
+            # flag-level silence beats Deprecations error
+            (["Deprecations"], [], ["BehaviorChangeEvent:use_mat_v2"], "use_mat_v2", False),
+            # different flag name — flag-level error does not fire
+            (["BehaviorChangeEvent:other_flag"], [], [], "use_mat_v2", False),
+            # flag-level warn overrides class-level error
+            (["BehaviorChangeEvent"], ["BehaviorChangeEvent:use_mat_v2"], [], "use_mat_v2", False),
+            # same flag in both error and silence: silence wins (silence > warn > error)
+            (
+                ["BehaviorChangeEvent:use_mat_v2"],
+                [],
+                ["BehaviorChangeEvent:use_mat_v2"],
+                "use_mat_v2",
+                False,
+            ),
+        ],
+    )
+    def test_errors_per_flag(
+        self,
+        error: Union[str, List[str]],
+        warn: List[str],
+        silence: List[str],
+        flag_name: str,
+        expected_errors: bool,
+    ) -> None:
+        event = BehaviorChangeEvent(
+            flag_name=flag_name,
+            flag_source="dbt_common",
+            description="test",
+            docs_url="https://docs.getdbt.com",
+        )
+        error_warn = WarnErrorOptionsV2(
+            error=error,
+            warn=warn,
+            silence=silence,
+            valid_error_names={"BehaviorChangeEvent", "ItemB"},
+        )
+        assert error_warn.errors(event) == expected_errors
+
+    # Per-flag BehaviorChange targeting in silenced()
+    @pytest.mark.parametrize(
+        "error,warn,silence,flag_name,expected_silence",
+        [
+            # primary use case: flag-level silence fires
+            ([], [], ["BehaviorChangeEvent:use_mat_v2"], "use_mat_v2", True),
+            # flag-level error beats class-level silence
+            (["BehaviorChangeEvent:use_mat_v2"], [], ["BehaviorChangeEvent"], "use_mat_v2", False),
+            # flag-level silence beats class-level error
+            (["BehaviorChangeEvent"], [], ["BehaviorChangeEvent:use_mat_v2"], "use_mat_v2", True),
+            # different flag not silenced
+            ([], [], ["BehaviorChangeEvent:other_flag"], "use_mat_v2", False),
+            # same flag in both error and silence: silence wins (silence > warn > error)
+            (
+                ["BehaviorChangeEvent:use_mat_v2"],
+                [],
+                ["BehaviorChangeEvent:use_mat_v2"],
+                "use_mat_v2",
+                True,
+            ),
+        ],
+    )
+    def test_silenced_per_flag(
+        self,
+        error: Union[str, List[str]],
+        warn: List[str],
+        silence: List[str],
+        flag_name: str,
+        expected_silence: bool,
+    ) -> None:
+        event = BehaviorChangeEvent(
+            flag_name=flag_name,
+            flag_source="dbt_common",
+            description="test",
+            docs_url="https://docs.getdbt.com",
+        )
+        my_options = WarnErrorOptionsV2(
+            error=error,
+            warn=warn,
+            silence=silence,
+            valid_error_names={"BehaviorChangeEvent", "ItemB"},
+        )
+        assert my_options.silenced(event) == expected_silence