Process review comments

src/dbt_score/__init__.py

@@ -1,11 +1 @@
 """Init dbt_score package."""
-
-import os
-from pathlib import Path
-
-MANIFEST_PATH = (
-    Path()
-    / os.getenv("DBT_PROJECT_DIR", "")
-    / os.getenv("DBT_TARGET_DIR", "target")
-    / "manifest.json"
-)

src/dbt_score/__main__.py

@@ -12,10 +12,10 @@
 def set_logging() -> None:
     """Set logging configuration."""
     log_format = "%(asctime)s %(levelname)s [%(name)s] %(message)s"
-    handlers: list[logging.Handler] = [logging.StreamHandler(sys.stdout)]
+    handler: logging.Handler = logging.StreamHandler(sys.stdout)
     logging.basicConfig(
         format=log_format,
-        handlers=handlers,
+        handlers=[handler],
         level=logging.INFO,
     )
     for handler in logging.getLogger().handlers:

src/dbt_score/cli.py

@@ -1,14 +1,14 @@
 """CLI interface."""
 
 import logging
-import pathlib
+from pathlib import Path
 from typing import Final
 
 import click
 from dbt.cli.options import MultiOption
 
-from dbt_score import MANIFEST_PATH
-from dbt_score.dbt_utils import dbt_parse
+from dbt_score.dbt_utils import get_manifest_path
+from dbt_score.lint import lint_manifest
 
 logger = logging.getLogger(__name__)
 
@@ -42,9 +42,7 @@ def cli() -> None:
     help="Manifest filepath.",
     type=click.Path(exists=True),
 )
-def lint(select: tuple[str], manifest: pathlib.Path) -> None:
+def lint(select: tuple[str], manifest: Path) -> None:
     """Lint dbt models metadata."""
-    manifest_path = MANIFEST_PATH if not manifest else manifest
-    if not manifest_path.exists():
-        logger.info("Executing dbt parse.")
-        dbt_parse()
+    manifest_path = get_manifest_path() if not manifest else manifest
+    lint_manifest(manifest_path)

src/dbt_score/dbt_utils.py

@@ -1,4 +1,6 @@
 """dbt utilities."""
+import os
+from pathlib import Path
 
 from dbt.cli.main import dbtRunner, dbtRunnerResult
 
@@ -12,12 +14,22 @@ def dbt_parse() -> dbtRunnerResult:
 
     Returns:
         dbtRunnerResult: dbt parse result
+
     raises:
         DbtParseException: dbt parse failed
     """
     result: dbtRunnerResult = dbtRunner().invoke(["parse"])
-
     if not result.success:
         raise DbtParseException("dbt parse failed.") from result.exception
 
     return result
+
+
+def get_manifest_path() -> Path:
+    """Get the manifest path."""
+    return (
+        Path().cwd()
+        / os.getenv("DBT_PROJECT_DIR", "")
+        / os.getenv("DBT_TARGET_DIR", "target")
+        / "manifest.json"
+    )

src/dbt_score/lint.py

@@ -0,0 +1,15 @@
+"""Lint dbt models metadata."""
+
+import logging
+from pathlib import Path
+
+from dbt_score.dbt_utils import dbt_parse
+
+logger = logging.getLogger(__name__)
+
+
+def lint_manifest(manifest_path: Path) -> None:
+    """Lint dbt manifest."""
+    if not manifest_path.exists():
+        logger.info("Executing dbt parse.")
+        dbt_parse()

src/dbt_score/rule_registry.py

@@ -4,12 +4,15 @@
 """
 
 import importlib
+import logging
 import pkgutil
 from typing import Iterator, Type
 
 from dbt_score.exceptions import DuplicatedRuleException
 from dbt_score.rule import Rule
 
+logger = logging.getLogger(__name__)
+
 THIRD_PARTY_RULES_NAMESPACE = "dbt_score_rules"
 
 
@@ -33,7 +36,7 @@ def _walk_packages(self, namespace_name: str) -> Iterator[str]:
             return
 
         def onerror(module_name: str) -> None:
-            print(f"Failed to import {module_name}.")
+            logger.warning(f"Failed to import {module_name}.")
 
         for package in pkgutil.walk_packages(namespace.__path__, onerror=onerror):
             yield f"{namespace_name}.{package.name}"

tests/test_cli.py

@@ -1,19 +1,31 @@
 """Test the CLI."""
-
+from pathlib import Path
 from unittest.mock import patch
 
 from click.testing import CliRunner
 from dbt_score.cli import lint
 
 
-@patch("dbt_score.cli.dbt_parse")
-def test_parse(dbt_parse, manifest_path):
+@patch("dbt_score.lint.dbt_parse")
+def test_parse_not_called(dbt_parse, manifest_path):
     """Test to ensure dbt parse is not called when the manifest exists."""
-    with patch("dbt_score.cli.MANIFEST_PATH", manifest_path):
-        runner = CliRunner()
+    runner = CliRunner()
+    with patch("dbt_score.cli.get_manifest_path", return_value=manifest_path):
         result1 = runner.invoke(lint)
-        result2 = runner.invoke(lint, ["--manifest", manifest_path])
+    result2 = runner.invoke(lint, ["--manifest", manifest_path])
+
+    assert result1.exit_code == 0
+    assert result2.exit_code == 0
+    dbt_parse.assert_not_called()
+
+
+@patch("dbt_score.lint.dbt_parse")
+def test_parse_called(dbt_parse):
+    """Test to ensure dbt parse is called when the manifest does not exist."""
+    runner = CliRunner()
+    with patch(
+        "dbt_score.cli.get_manifest_path", return_value=Path("fake_manifest.json")
+    ):
+        runner.invoke(lint)
 
-        assert result1.exit_code == 0
-        assert result2.exit_code == 0
-        dbt_parse.assert_not_called()
+    dbt_parse.assert_called_once()