Add new-syle YAML unparsed metric

Author: theyostalservice

.changes/unreleased/Features-20251111-103504.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add UnparsedMetricV2 to read in new-style YAML Semantic Layer Metrics.
+time: 2025-11-11T10:35:04.123144-08:00
+custom:
+    Author: theyostalservice
+    Issue: na

core/dbt/contracts/graph/unparsed.py

@@ -610,24 +610,10 @@ class UnparsedMetricTypeParams(dbtClassMixin):
     cumulative_type_params: Optional[UnparsedCumulativeTypeParams] = None
 
 
-@dataclass
-class UnparsedMetric(dbtClassMixin):
-    name: str
-    label: str
-    type: str
-    type_params: UnparsedMetricTypeParams
-    description: str = ""
-    # Note: `Union` must be the outermost part of the type annotation for serialization to work properly.
-    filter: Union[str, List[str], None] = None
-    time_granularity: Optional[str] = None
-    # metadata: Optional[Unparsedetadata] = None # TODO
-    meta: Dict[str, Any] = field(default_factory=dict)
-    tags: List[str] = field(default_factory=list)
-    config: Dict[str, Any] = field(default_factory=dict)
-
+class UnparsedMetricBase(dbtClassMixin):
     @classmethod
     def validate(cls, data):
-        super(UnparsedMetric, cls).validate(data)
+        super().validate(data)
         if "name" in data:
             errors = []
             if " " in data["name"]:
@@ -647,6 +633,76 @@ def validate(cls, data):
                 )
 
 
+@dataclass
+class UnparsedMetric(UnparsedMetricBase):
+    """Old-style YAML metric; prefer UnparsedMetricV2 instead as of late 2025."""
+
+    name: str
+    label: str
+    type: str
+    type_params: UnparsedMetricTypeParams  # old-style YAML
+    description: str = ""
+    # Note: `Union` must be the outermost part of the type annotation for serialization to work properly.
+    filter: Union[str, List[str], None] = None
+    time_granularity: Optional[str] = None
+    # metadata: Optional[Unparsedetadata] = None # TODO
+    meta: Dict[str, Any] = field(default_factory=dict)
+    tags: List[str] = field(default_factory=list)
+    config: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class UnparsedNonAdditiveDimensionV2(dbtClassMixin):
+    name: str
+    window_agg: str  # AggregationType enum
+    group_by: List[str] = field(default_factory=list)
+
+
+@dataclass
+class UnparsedMetricV2(UnparsedMetricBase):
+    name: str
+    label: Optional[str] = None
+    hidden: bool = False
+    description: Optional[str] = None
+    type: Optional[str] = "simple"
+    agg: Optional[str] = None
+
+    percentile: Optional[float] = None
+    percentile_type: Optional[str] = None
+
+    join_to_timespine: Optional[bool] = None
+    fill_nulls_with: Optional[int] = None
+    expr: Optional[Union[str, int]] = None
+    filter: Union[str, List[str], None] = None
+
+    tags: List[str] = field(default_factory=list)
+    meta: Dict[str, Any] = field(default_factory=dict)
+    config: Dict[str, Any] = field(default_factory=dict)
+
+    non_additive_dimension: Optional[UnparsedNonAdditiveDimensionV2] = None
+    agg_time_dimension: Optional[str] = None
+
+    # For cumulative metrics
+    window: Optional[str] = None
+    grain_to_date: Optional[str] = None
+    period_agg: Optional[str] = None
+    input_metric: Optional[Union[str, Dict[str, Any]]] = None
+
+    # For ratio metrics
+    numerator: Optional[Union[str, Dict[str, Any]]] = None
+    denominator: Optional[Union[str, Dict[str, Any]]] = None
+
+    # For derived metrics
+    input_metrics: Optional[List[Dict[str, Any]]] = None
+
+    # For conversion metrics
+    entity: Optional[str] = None
+    calculation: Optional[str] = None
+    base_metric: Optional[Union[str, Dict[str, Any]]] = None
+    conversion_metric: Optional[Union[str, Dict[str, Any]]] = None
+    constant_properties: Optional[List[Dict[str, Any]]] = None
+
+
 @dataclass
 class UnparsedGroup(dbtClassMixin):
     name: str

tests/unit/contracts/graph/test_unparsed.py

@@ -1,7 +1,10 @@
 import pickle
+from abc import abstractmethod
 from datetime import timedelta
+from typing import Any, Dict
 
 import pytest
+from typing_extensions import override
 
 from dbt.artifacts.resources import (
     ExposureType,
@@ -23,6 +26,7 @@
     UnparsedMetric,
     UnparsedMetricInputMeasure,
     UnparsedMetricTypeParams,
+    UnparsedMetricV2,
     UnparsedModelUpdate,
     UnparsedNode,
     UnparsedNodeUpdate,
@@ -883,9 +887,57 @@ def test_bad_tags(self):
         self.assert_fails_validation(tst)
 
 
-class TestUnparsedMetric(ContractTestCase):
+class BaseTestUnparsedMetric:
+
+    @abstractmethod
+    def get_ok_dict(self) -> Dict[str, Any]:
+        raise NotImplementedError()
+
+    def test_bad_tags(self):
+        tst = self.get_ok_dict()
+        tst["tags"] = [123]
+        self.assert_fails_validation(tst)
+
+    def test_bad_metric_name_with_spaces(self):
+        tst = self.get_ok_dict()
+        tst["name"] = "metric name with spaces"
+        self.assert_fails_validation(tst)
+
+    def test_bad_metric_name_too_long(self):
+        tst = self.get_ok_dict()
+        tst["name"] = "a" * 251
+        self.assert_fails_validation(tst)
+
+    def test_bad_metric_name_does_not_start_with_letter(self):
+        tst = self.get_ok_dict()
+        tst["name"] = "123metric"
+        self.assert_fails_validation(tst)
+
+        tst["name"] = "_metric"
+        self.assert_fails_validation(tst)
+
+    def test_bad_metric_name_contains_special_characters(self):
+        tst = self.get_ok_dict()
+        tst["name"] = "metric!name"
+        self.assert_fails_validation(tst)
+
+        tst["name"] = "metric@name"
+        self.assert_fails_validation(tst)
+
+        tst["name"] = "metric#name"
+        self.assert_fails_validation(tst)
+
+        tst["name"] = "metric$name"
+        self.assert_fails_validation(tst)
+
+        tst["name"] = "metric-name"
+        self.assert_fails_validation(tst)
+
+
+class TestUnparsedMetric(BaseTestUnparsedMetric, ContractTestCase):
     ContractType = UnparsedMetric
 
+    @override
     def get_ok_dict(self):
         return {
             "name": "new_customers",
@@ -928,45 +980,42 @@ def test_bad_metric_no_type_params(self):
         del tst["type_params"]
         self.assert_fails_validation(tst)
 
-    def test_bad_tags(self):
-        tst = self.get_ok_dict()
-        tst["tags"] = [123]
-        self.assert_fails_validation(tst)
-
-    def test_bad_metric_name_with_spaces(self):
-        tst = self.get_ok_dict()
-        tst["name"] = "metric name with spaces"
-        self.assert_fails_validation(tst)
-
-    def test_bad_metric_name_too_long(self):
-        tst = self.get_ok_dict()
-        tst["name"] = "a" * 251
-        self.assert_fails_validation(tst)
 
-    def test_bad_metric_name_does_not_start_with_letter(self):
-        tst = self.get_ok_dict()
-        tst["name"] = "123metric"
-        self.assert_fails_validation(tst)
-
-        tst["name"] = "_metric"
-        self.assert_fails_validation(tst)
-
-    def test_bad_metric_name_contains_special_characters(self):
-        tst = self.get_ok_dict()
-        tst["name"] = "metric!name"
-        self.assert_fails_validation(tst)
-
-        tst["name"] = "metric@name"
-        self.assert_fails_validation(tst)
-
-        tst["name"] = "metric#name"
-        self.assert_fails_validation(tst)
+class TestUnparsedMetricV2(BaseTestUnparsedMetric, ContractTestCase):
+    ContractType = UnparsedMetricV2
 
-        tst["name"] = "metric$name"
-        self.assert_fails_validation(tst)
+    @override
+    def get_ok_dict(self):
+        return {
+            "name": "new_customers",
+            "label": "New Customers",
+            "description": "New customers",
+            "type": "simple",
+            "agg": "sum",
+            "filter": "is_new = true",
+            "join_to_timespine": False,
+            "config": {},
+            "tags": [],
+            "meta": {"is_okr": True},
+        }
 
-        tst["name"] = "metric-name"
-        self.assert_fails_validation(tst)
+    def test_ok(self):
+        metric = self.ContractType(
+            name="new_customers",
+            label="New Customers",
+            description="New customers",
+            agg="sum",
+            filter="is_new = true",
+            join_to_timespine=False,
+            config={},
+            tags=[],
+            meta={"is_okr": True},
+        )
+        dct = self.get_ok_dict()
+        # add defaults:
+        dct["hidden"] = False
+        self.assert_symmetric(metric, dct)
+        pickle.loads(pickle.dumps(metric))
 
 
 class TestUnparsedVersion(ContractTestCase):