add: sqlmesh key_metrics model

warehouse/metrics_tools/definition.py

@@ -433,6 +433,17 @@ def generate_dependency_refs_for_name(self, name: str):
                         time_aggregation=time_aggregation,
                     )
                 )
+            # if there is no _source.time_aggregations or _source.rolling
+            # means it is a point in time metric aka key metric over the
+            # whole time period for the specific entity type
+            if not self._source.time_aggregations and not self._source.rolling:
+                refs.append(
+                    PeerMetricDependencyRef(
+                        name=name,
+                        entity_type=entity,
+                        time_aggregation="overall",
+                    )
+                )
         return refs
 
     @property
@@ -449,8 +460,9 @@ def metric_type(self):
             return "time_aggregation"
         elif self._source.rolling is not None:
             return "rolling"
-        # This _shouldn't_ happen
-        raise Exception("unknown metric type")
+        # If neither time_aggregations or rolling is set then it is a point in
+        # time metric
+        return "overall"
 
     def generate_query_ref(
         self,

warehouse/metrics_tools/factory/factory.py

@@ -393,6 +393,10 @@ def generate_model_for_rendered_query(
                 self.generate_time_aggregation_model_for_rendered_query(
                     calling_file, query_config, dependencies
                 )
+            case "overall":
+                self.generate_point_in_time_model_for_rendered_query(
+                    calling_file, query_config, dependencies
+                )
 
     def generate_rolling_python_model_for_rendered_query(
         self,
@@ -546,6 +550,40 @@ def generate_time_aggregation_model_for_rendered_query(
             partitioned_by=partitioned_by,
         )
 
+    def generate_point_in_time_model_for_rendered_query(
+        self,
+        calling_file: str,
+        query_config: MetricQueryConfig,
+        dependencies: t.Set[str],
+    ):
+        """Generate model for point in time models"""
+        config = self.serializable_config(query_config)
+        ref = query_config["ref"]
+        columns = METRICS_COLUMNS_BY_ENTITY[ref["entity_type"]]
+
+        grain = [
+            "metric",
+            f"to_{ref['entity_type']}_id",
+            "from_artifact_id",
+            "event_source",
+            "metrics_sample_date",
+        ]
+
+        GeneratedModel.create(
+            func=generated_query,
+            entrypoint_path=calling_file,
+            config=config,
+            name=f"{self.catalog}.{query_config['table_name']}",
+            kind=ModelKindName.VIEW,
+            dialect="clickhouse",
+            columns=columns,
+            grain=grain,
+            start=self._raw_options["start"],
+            # since this query's over the full time range, we don't expose access to the
+            # macros used to get information about the specific time
+            additional_macros=[],
+        )
+
     def serializable_config(self, query_config: MetricQueryConfig):
         # Use a simple python sql model to generate the time_aggregation model
         config: t.Dict[str, t.Any] = t.cast(dict, query_config.copy())