Add new-syle YAML metric

Author: theyostalservice

core/dbt/contracts/graph/unparsed.py

@@ -610,12 +610,37 @@ class UnparsedMetricTypeParams(dbtClassMixin):
     cumulative_type_params: Optional[UnparsedCumulativeTypeParams] = None
 
 
+class UnparsedMetricBase(dbtClassMixin):
+    @classmethod
+    def validate(cls, data):
+        super().validate(data)
+        if "name" in data:
+            errors = []
+            if " " in data["name"]:
+                errors.append("cannot contain spaces")
+            # This handles failing queries due to too long metric names.
+            # It only occurs in BigQuery and Snowflake (Postgres/Redshift truncate)
+            if len(data["name"]) > 250:
+                errors.append("cannot contain more than 250 characters")
+            if not (re.match(r"^[A-Za-z]", data["name"])):
+                errors.append("must begin with a letter")
+            if not (re.match(r"[\w-]+$", data["name"])):
+                errors.append("must contain only letters, numbers and underscores")
+
+            if errors:
+                raise ParsingError(
+                    f"The metric name '{data['name']}' is invalid.  It {', '.join(e for e in errors)}"
+                )
+
+
 @dataclass
-class UnparsedMetric(dbtClassMixin):
+class UnparsedMetric(UnparsedMetricBase):
+    """Old-style YAML metric; prefer UnparsedMetricV2 instead as of late 2025."""
+
     name: str
     label: str
     type: str
-    type_params: UnparsedMetricTypeParams
+    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
@@ -625,22 +650,71 @@ class UnparsedMetric(dbtClassMixin):
     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
+
     @classmethod
     def validate(cls, data):
-        super(UnparsedMetric, cls).validate(data)
+        super(UnparsedMetricV2, cls).validate(data)
         if "name" in data:
             errors = []
             if " " in data["name"]:
                 errors.append("cannot contain spaces")
-            # This handles failing queries due to too long metric names.
-            # It only occurs in BigQuery and Snowflake (Postgres/Redshift truncate)
             if len(data["name"]) > 250:
                 errors.append("cannot contain more than 250 characters")
             if not (re.match(r"^[A-Za-z]", data["name"])):
                 errors.append("must begin with a letter")
             if not (re.match(r"[\w-]+$", data["name"])):
                 errors.append("must contain only letters, numbers and underscores")
-
             if errors:
                 raise ParsingError(
                     f"The metric name '{data['name']}' is invalid.  It {', '.join(e for e in errors)}"

core/dbt/parser/schema_yaml_readers.py

@@ -72,7 +72,7 @@
 
 
 def parse_where_filter(
-    where: Optional[Union[List[str], str]]
+    where: Optional[Union[List[str], str]],
 ) -> Optional[WhereFilterIntersection]:
     if where is None:
         return None
@@ -470,12 +470,19 @@ def _generate_metric_config(
 
     def parse(self) -> None:
         for data in self.get_key_dicts():
-            try:
-                UnparsedMetric.validate(data)
-                unparsed = UnparsedMetric.from_dict(data)
-
-            except (ValidationError, JSONValidationError) as exc:
-                raise YamlParseDictError(self.yaml.path, self.key, data, exc)
+            if "type_params" in data:
+                try:
+                    UnparsedMetric.validate(data)
+                    unparsed = UnparsedMetric.from_dict(data)
+
+                except (ValidationError, JSONValidationError) as exc:
+                    raise YamlParseDictError(self.yaml.path, self.key, data, exc)
+            else:
+                try:
+                    UnparsedMetricV2.validate(data)
+                    unparsed = UnparsedMetricV2.from_dict(data)
+                except (ValidationError, JSONValidationError) as exc:
+                    raise YamlParseDictError(self.yaml.path, self.key, data, exc)
             self.parse_metric(unparsed)