Merge pull request #264 from godatadriven/rearrange-parsers

Rearrange parsers

pyproject.toml

@@ -110,7 +110,9 @@ pytest-xdist = "^3"
 ruff = "^0"
 sh = "^2"
 "shandy-sqlfmt" = {version = "^0", extras=["shandy-sqlfmt[jinjafmt]"]}
+types-PyYaml = "^6"
 types-tabulate = "^0"
+types-toml = "^0"
 
 [tool.poetry.group.docs.dependencies]
 mike = "^2"
@@ -154,7 +156,7 @@ exclude = [
     "dist",
     "node_modules",
     "site-packages",
-    "src/dbt_bouncer/dbt_cloud",
+    "src/dbt_bouncer/artifact_parsers/dbt_cloud",
     "venv",
 ]
 line-length = 88

src/dbt_bouncer/dbt_cloud/manifest_latest.py → src/dbt_bouncer/artifact_parsers/dbt_cloud/manifest_latest.py

@@ -15,15 +15,15 @@
 
 class CustomGranularity(BaseParserModel):
     model_config = ConfigDict(
-        extra='forbid',
+        extra="forbid",
     )
     name: str
     column_name: Optional[str] = None
 
 
 class TimeSpine(BaseParserModel):
     model_config = ConfigDict(
-        extra='forbid',
+        extra="forbid",
     )
     standard_granularity_column: str
     custom_granularities: Optional[List[CustomGranularity]] = None

src/dbt_bouncer/artifact_parsers/parsers_catalog.py

@@ -0,0 +1,80 @@
+# TODO Remove after this program no longer support Python 3.8.*
+from __future__ import annotations
+
+import logging
+import warnings
+from typing import TYPE_CHECKING, List
+
+from pydantic import BaseModel
+
+from dbt_bouncer.utils import clean_path_str
+
+with warnings.catch_warnings():
+    warnings.filterwarnings("ignore", category=UserWarning)
+    from dbt_artifacts_parser.parsers.catalog.catalog_v1 import (
+        CatalogTable,  # noqa: TCH002
+        CatalogV1,  # noqa: TCH002
+    )
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from dbt_bouncer.artifact_parsers.parsers_manifest import DbtBouncerManifest
+
+
+from dbt_bouncer.artifact_parsers.parsers_common import load_dbt_artifact
+
+
+class DbtBouncerCatalogNode(BaseModel):
+    """Model for all nodes in `catalog.json`."""
+
+    catalog_node: CatalogTable
+    original_file_path: str
+    unique_id: str
+
+
+def parse_catalog(
+    artifact_dir: Path,
+    manifest_obj: "DbtBouncerManifest",
+) -> tuple[List[DbtBouncerCatalogNode], List[DbtBouncerCatalogNode]]:
+    """Parse the catalog.json artifact.
+
+    Returns:
+        List[DbtBouncerCatalogNode]: List of catalog nodes for the project.
+        List[DbtBouncerCatalogNode]: List of catalog nodes for the project sources.
+
+    """
+    catalog_obj: CatalogV1 = load_dbt_artifact(
+        artifact_name="catalog.json",
+        dbt_artifacts_dir=artifact_dir,
+    )
+    project_catalog_nodes = [
+        DbtBouncerCatalogNode(
+            **{
+                "catalog_node": v,
+                "original_file_path": clean_path_str(
+                    manifest_obj.manifest.nodes[k].original_file_path
+                ),
+                "unique_id": k,
+            },
+        )
+        for k, v in catalog_obj.nodes.items()
+        if k.split(".")[-2] == manifest_obj.manifest.metadata.project_name
+    ]
+    project_catalog_sources = [
+        DbtBouncerCatalogNode(
+            **{
+                "catalog_node": v,
+                "original_file_path": clean_path_str(
+                    manifest_obj.manifest.sources[k].original_file_path
+                ),
+                "unique_id": k,
+            },
+        )
+        for k, v in catalog_obj.sources.items()
+        if k.split(".")[1] == manifest_obj.manifest.metadata.project_name
+    ]
+    logging.info(
+        f"Parsed `catalog.json`: {len(project_catalog_nodes)} nodes, {len(project_catalog_sources)} sources.",
+    )
+
+    return project_catalog_nodes, project_catalog_sources