added data assets group to the tree

elementary/monitor/api/groups/groups.py

@@ -9,6 +9,7 @@
     OwnersGroupSchema,
     TagsGroupSchema,
     TreeGroupSchema,
+    TreeViewSchema,
 )
 from elementary.monitor.api.groups.tree_builder import TreeBuilder
 from elementary.monitor.api.models.schema import (
@@ -35,41 +36,79 @@
 
 class GroupsAPI(APIClient):
     def get_groups(self, artifacts: List[GROUPABLE_ARTIFACT]) -> GroupsSchema:
-        dbt_group = self.get_dbt_group(artifacts)
-        tags_group = self.get_tags_group(artifacts)
-        owners_group = self.get_owners_group(artifacts)
-        dwh_group = self.get_dwh_group(artifacts)
+        data_assets_group = self.get_data_assets_group(artifacts)
+
+        # direct views - deprecated
+        dbt_group = self.get_normalized_full_path_view(artifacts)
+        tags_group = self.get_tags_view(artifacts)
+        owners_group = self.get_owners_view(artifacts)
+        dwh_group = self.get_fqn_view(artifacts)
+
         return GroupsSchema(
-            dbt=dbt_group, dwh=dwh_group, tags=tags_group, owners=owners_group
+            dbt=dbt_group,
+            dwh=dwh_group,
+            tags=tags_group,
+            owners=owners_group,
+            data_assets=data_assets_group,
         )
 
-    def get_dbt_group(
+    def get_data_assets_group(
+        self, artifacts: List[GROUPABLE_ARTIFACT]
+    ) -> List[TreeViewSchema]:
+        filtered_artifacts = self.filter_data_assets_artifacts(artifacts)
+
+        dbt_view = self.get_normalized_full_path_view(filtered_artifacts)
+        tags_view = self.get_tags_view(filtered_artifacts)
+        owners_view = self.get_owners_view(filtered_artifacts)
+        dwh_view = self.get_fqn_view(filtered_artifacts)
+
+        return [
+            TreeViewSchema(name="dbt", data=dbt_view),
+            TreeViewSchema(name="tags", data=tags_view),
+            TreeViewSchema(name="owners", data=owners_view),
+            TreeViewSchema(name="dwh", data=dwh_view),
+        ]
+
+    @staticmethod
+    def filter_data_assets_artifacts(
+        artifacts: List[GROUPABLE_ARTIFACT],
+    ) -> List[GROUPABLE_ARTIFACT]:
+        return [
+            artifact
+            for artifact in artifacts
+            if not isinstance(artifact, NormalizedExposureSchema)
+            or not artifact.meta
+            or artifact.meta.get("platform")
+        ]
+
+    def get_normalized_full_path_view(
         self,
         artifacts: List[GROUPABLE_ARTIFACT],
     ) -> TreeGroupSchema:
+        filtered_artifacts = [
+            artifact for artifact in artifacts if artifact.unique_id is not None
+        ]
         tree_builder = TreeBuilder[GroupItemSchema](separator=posixpath.sep)
-        for artifact in artifacts:
-            if artifact.unique_id is None:
-                continue
+        for artifact in filtered_artifacts:
             tree_builder.add(
                 path=artifact.normalized_full_path, data=self._get_group_item(artifact)
             )
         return tree_builder.get_tree()
 
-    def get_dwh_group(self, artifacts: List[GROUPABLE_ARTIFACT]) -> TreeGroupSchema:
-        tree_builder = TreeBuilder[GroupItemSchema](separator=".")
-        relevant_artifacts = (
+    def get_fqn_view(self, artifacts: List[GROUPABLE_ARTIFACT]) -> TreeGroupSchema:
+        filtered_artifacts = (
             artifact
             for artifact in artifacts
             if artifact.unique_id is not None
             and artifact.fqn is not None
             and isinstance(artifact, (NormalizedSourceSchema, NormalizedModelSchema))
         )
-        for artifact in relevant_artifacts:
+        tree_builder = TreeBuilder[GroupItemSchema](separator=".")
+        for artifact in filtered_artifacts:
             tree_builder.add(path=artifact.fqn, data=self._get_group_item(artifact))
         return tree_builder.get_tree()
 
-    def get_tags_group(
+    def get_tags_view(
         self,
         artifacts: List[GROUPABLE_ARTIFACT],
     ) -> TagsGroupSchema:
@@ -86,9 +125,7 @@ def get_tags_group(
                 group[NO_TAGS_DEFAULT_TREE].append(self._get_group_item(artifact))
         return dict(group)
 
-    def get_owners_group(
-        self, artifacts: List[GROUPABLE_ARTIFACT]
-    ) -> OwnersGroupSchema:
+    def get_owners_view(self, artifacts: List[GROUPABLE_ARTIFACT]) -> OwnersGroupSchema:
         group = defaultdict(list)
         for artifact in artifacts:
             unique_id = artifact.unique_id

elementary/monitor/api/groups/schema.py

@@ -13,7 +13,16 @@ class GroupItemSchema(BaseModel):
 OwnersGroupSchema = Dict[str, List[GroupItemSchema]]
 
 
+class TreeViewSchema(BaseModel):
+    name: str
+    data: Dict[str, Any]
+
+
 class GroupsSchema(BaseModel):
+    data_assets: List[TreeViewSchema] = []
+    bi_assets: Optional[List[TreeViewSchema]] = None
+
+    # deprecated
     dbt: TreeGroupSchema = dict()
     tags: TagsGroupSchema = dict()
     owners: OwnersGroupSchema = dict()