CT-2752: Extract methods to new SemanticManifest class for better enc…

…apsulation of details. (#8012)

core/dbt/contracts/graph/manifest.py

@@ -50,7 +50,6 @@
     DuplicateMacroInPackageError,
     DuplicateMaterializationNameError,
     AmbiguousResourceNameRefError,
-    ParsingError,
 )
 from dbt.helper_types import PathSet
 from dbt.events.functions import fire_event
@@ -60,16 +59,7 @@
 from dbt.flags import get_flags, MP_CONTEXT
 from dbt import tracking
 import dbt.utils
-from dbt_semantic_interfaces.implementations.metric import PydanticMetric
-from dbt_semantic_interfaces.implementations.semantic_manifest import PydanticSemanticManifest
-from dbt_semantic_interfaces.implementations.semantic_model import PydanticSemanticModel
-from dbt_semantic_interfaces.implementations.project_configuration import (
-    PydanticProjectConfiguration,
-)
-from dbt_semantic_interfaces.implementations.time_spine_table_configuration import (
-    PydanticTimeSpineTableConfiguration,
-)
-from dbt_semantic_interfaces.type_enums import TimeGranularity
+
 
 NodeEdgeMap = Dict[str, List[str]]
 PackageName = str
@@ -988,44 +978,6 @@ def analysis_lookup(self) -> AnalysisLookup:
             self._analysis_lookup = AnalysisLookup(self)
         return self._analysis_lookup
 
-    @property
-    def pydantic_semantic_manifest(self) -> PydanticSemanticManifest:
-        project_config = PydanticProjectConfiguration(
-            time_spine_table_configurations=[],
-        )
-        pydantic_semantic_manifest = PydanticSemanticManifest(
-            metrics=[], semantic_models=[], project_configuration=project_config
-        )
-
-        for semantic_model in self.semantic_models.values():
-            pydantic_semantic_manifest.semantic_models.append(
-                PydanticSemanticModel.parse_obj(semantic_model.to_dict())
-            )
-
-        for metric in self.metrics.values():
-            pydantic_semantic_manifest.metrics.append(PydanticMetric.parse_obj(metric.to_dict()))
-
-        # Look for time-spine table model and create time spine table configuration
-        if self.semantic_models:
-            # Get model for time_spine_table
-            time_spine_model_name = "metricflow_time_spine"
-            model = self.ref_lookup.find(time_spine_model_name, None, None, self)
-            if not model:
-                raise ParsingError(
-                    "The semantic layer requires a 'metricflow_time_spine' model in the project, but none was found. Guidance on creating this model can be found on our docs site (https://docs.getdbt.com/docs/build/metricflow-time-spine)"
-                )
-            # Create time_spine_table_config, set it in project_config, and add to semantic manifest
-            time_spine_table_config = PydanticTimeSpineTableConfiguration(
-                location=model.relation_name,
-                column_name="date_day",
-                grain=TimeGranularity.DAY,
-            )
-            pydantic_semantic_manifest.project_configuration.time_spine_table_configurations = [
-                time_spine_table_config
-            ]
-
-        return pydantic_semantic_manifest
-
     @property
     def external_node_unique_ids(self):
         return [node.unique_id for node in self.nodes.values() if node.is_external_node]

core/dbt/contracts/graph/semantic_manifest.py

@@ -0,0 +1,62 @@
+from dbt_semantic_interfaces.implementations.metric import PydanticMetric
+from dbt_semantic_interfaces.implementations.project_configuration import (
+    PydanticProjectConfiguration,
+)
+from dbt_semantic_interfaces.implementations.semantic_manifest import PydanticSemanticManifest
+from dbt_semantic_interfaces.implementations.semantic_model import PydanticSemanticModel
+from dbt_semantic_interfaces.implementations.time_spine_table_configuration import (
+    PydanticTimeSpineTableConfiguration,
+)
+from dbt_semantic_interfaces.type_enums import TimeGranularity
+
+from dbt.clients.system import write_file
+from dbt.exceptions import ParsingError
+
+
+class SemanticManifest:
+    def __init__(self, manifest):
+        self.manifest = manifest
+
+    def write_json_to_file(self, file_path: str):
+        semantic_manifest = self._get_pydantic_semantic_manifest()
+        json = semantic_manifest.json()
+        write_file(file_path, json)
+
+    def _get_pydantic_semantic_manifest(self) -> PydanticSemanticManifest:
+        project_config = PydanticProjectConfiguration(
+            time_spine_table_configurations=[],
+        )
+        pydantic_semantic_manifest = PydanticSemanticManifest(
+            metrics=[], semantic_models=[], project_configuration=project_config
+        )
+
+        for semantic_model in self.manifest.semantic_models.values():
+            pydantic_semantic_manifest.semantic_models.append(
+                PydanticSemanticModel.parse_obj(semantic_model.to_dict())
+            )
+
+        for metric in self.manifest.metrics.values():
+            pydantic_semantic_manifest.metrics.append(PydanticMetric.parse_obj(metric.to_dict()))
+
+        # Look for time-spine table model and create time spine table configuration
+        if self.manifest.semantic_models:
+            # Get model for time_spine_table
+            time_spine_model_name = "metricflow_time_spine"
+            model = self.manifest.ref_lookup.find(time_spine_model_name, None, None, self.manifest)
+            if not model:
+                raise ParsingError(
+                    "The semantic layer requires a 'metricflow_time_spine' model in the project, but none was found. "
+                    "Guidance on creating this model can be found on our docs site ("
+                    "https://docs.getdbt.com/docs/build/metricflow-time-spine) "
+                )
+            # Create time_spine_table_config, set it in project_config, and add to semantic manifest
+            time_spine_table_config = PydanticTimeSpineTableConfiguration(
+                location=model.relation_name,
+                column_name="date_day",
+                grain=TimeGranularity.DAY,
+            )
+            pydantic_semantic_manifest.project_configuration.time_spine_table_configurations = [
+                time_spine_table_config
+            ]
+
+        return pydantic_semantic_manifest

core/dbt/parser/manifest.py

@@ -18,6 +18,8 @@
 )
 from itertools import chain
 import time
+
+from dbt.contracts.graph.semantic_manifest import SemanticManifest
 from dbt.events.base_types import EventLevel
 import json
 import pprint
@@ -1651,7 +1653,8 @@ def process_node(config: RuntimeConfig, manifest: Manifest, node: ManifestNode):
 
 def write_semantic_manifest(manifest: Manifest, target_path: str) -> None:
     path = os.path.join(target_path, SEMANTIC_MANIFEST_FILE_NAME)
-    write_file(path, manifest.pydantic_semantic_manifest.json())
+    semantic_manifest = SemanticManifest(manifest)
+    semantic_manifest.write_json_to_file(path)
 
 
 def write_manifest(manifest: Manifest, target_path: str):