Add parsing and logging (#10)

Add optionally running `dbt parse` during `lint`.
- When a manifest is provided through `cli` (and it exists, because
otherwise it will raise an error), `dbt parse` is not executed.
- When a manifest exists in the dbt project, on its specified location,
`dbt parse` is not executed.

---------

Co-authored-by: Matthieu Caneill <[email protected]>

pyproject.toml

@@ -89,6 +89,11 @@ convention = "google"
 [tool.ruff.lint.pylint]
 max-args = 6
 
+[tool.ruff.lint.per-file-ignores]
+"tests/**/*.py" = [
+    "PLR2004",  # magic value comparisons
+]
+
 ### Coverage ###
 
 [tool.coverage.run]

src/dbt_score/__main__.py

@@ -3,7 +3,21 @@
 This enables module to be run directly.
 """
 
+import logging
+import sys
+
 from dbt_score.cli import cli
 
+
+def set_logging() -> None:
+    """Set logging configuration."""
+    log_format = "%(asctime)s %(levelname)s [%(name)s] %(message)s"
+    handler: logging.Handler = logging.StreamHandler(sys.stdout)
+    logging.basicConfig(format=log_format, handlers=[handler])
+    for handler in logging.getLogger().handlers:
+        handler.setLevel(logging.WARNING)
+
+
 if __name__ == "__main__":
+    set_logging()
     cli()

src/dbt_score/cli.py

@@ -1,10 +1,15 @@
 """CLI interface."""
 
+from pathlib import Path
 from typing import Final
 
 import click
+from click.core import ParameterSource
 from dbt.cli.options import MultiOption
 
+from dbt_score.lint import lint_dbt_project
+from dbt_score.parse import dbt_parse, get_default_manifest_path
+
 BANNER: Final[str] = r"""
           __ __     __
      ____/ // /_   / /_        _____ _____ ____   _____ ___
@@ -29,6 +34,30 @@ def cli() -> None:
     type=tuple,
     multiple=True,
 )
-def lint(select: tuple[str]) -> None:
+@click.option(
+    "--manifest",
+    "-m",
+    help="Manifest filepath.",
+    type=click.Path(path_type=Path),
+    default=get_default_manifest_path(),
+)
+@click.option(
+    "--run-dbt-parse",
+    "-p",
+    help="Run dbt parse.",
+    is_flag=True,
+    default=False,
+)
+def lint(select: tuple[str], manifest: Path, run_dbt_parse: bool) -> None:
     """Lint dbt models metadata."""
-    raise NotImplementedError()
+    manifest_provided = (
+        click.get_current_context().get_parameter_source("manifest")
+        != ParameterSource.DEFAULT
+    )
+    if manifest_provided and run_dbt_parse:
+        raise click.UsageError("--run-dbt-parse cannot be used with --manifest.")
+
+    if run_dbt_parse:
+        dbt_parse()
+
+    lint_dbt_project(manifest)

src/dbt_score/lint.py

@@ -0,0 +1,9 @@
+"""Lint dbt models metadata."""
+
+from pathlib import Path
+
+
+def lint_dbt_project(manifest_path: Path) -> None:
+    """Lint dbt manifest."""
+    if not manifest_path.exists():
+        raise FileNotFoundError(f"Manifest not found at {manifest_path}.")

src/dbt_score/parse.py

@@ -0,0 +1,43 @@
+"""dbt utilities."""
+
+import logging
+import os
+from pathlib import Path
+
+from dbt.cli.main import dbtRunner, dbtRunnerResult
+
+
+class DbtParseException(Exception):
+    """Raised when dbt parse fails."""
+
+
+def dbt_parse() -> dbtRunnerResult:
+    """Parse a dbt project.
+
+    Returns:
+        The dbt parse run result.
+
+    Raises:
+        DbtParseException: dbt parse failed.
+    """
+    dbt_logger_stdout = logging.getLogger("stdout_log")
+    dbt_logger_stdout.disabled = True
+
+    result: dbtRunnerResult = dbtRunner().invoke(["parse"])
+
+    dbt_logger_stdout.disabled = False
+
+    if not result.success:
+        raise DbtParseException("dbt parse failed.") from result.exception
+
+    return result
+
+
+def get_default_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/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/conftest.py

@@ -10,13 +10,15 @@
 
 
 @fixture
-def raw_manifest() -> Any:
+def manifest_path() -> Path:
+    """Mock the manifest path."""
+    return Path(__file__).parent / "resources" / "manifest.json"
+
+
+@fixture
+def raw_manifest(manifest_path) -> Any:
     """Mock the raw manifest."""
-    return json.loads(
-        (Path(__file__).parent / "resources" / "manifest.json").read_text(
-            encoding="utf-8"
-        )
-    )
+    return json.loads(manifest_path.read_text(encoding="utf-8"))
 
 
 @fixture

tests/test_cli.py

@@ -0,0 +1,36 @@
+"""Test the CLI."""
+
+import pytest
+from click.testing import CliRunner
+from dbt_score.cli import lint
+
+
+def test_invalid_options():
+    """Test invalid cli options."""
+    runner = CliRunner()
+    result = runner.invoke(
+        lint, ["--manifest", "fake_manifest.json", "--run-dbt-parse"]
+    )
+    assert result.exit_code == 2  # pylint: disable=PLR2004
+
+
+def test_lint_existing_manifest(manifest_path):
+    """Test lint with an existing manifest."""
+    runner = CliRunner()
+    result = runner.invoke(lint, ["--manifest", manifest_path])
+    assert result.exit_code == 0
+
+
+def test_lint_non_existing_manifest():
+    """Test lint with a non-existing manifest."""
+    runner = CliRunner()
+
+    # Provide manifest in command line.
+    with pytest.raises(FileNotFoundError):
+        runner.invoke(
+            lint, ["--manifest", "fake_manifest.json"], catch_exceptions=False
+        )
+
+    # Use default manifest path..
+    with pytest.raises(FileNotFoundError):
+        runner.invoke(lint, catch_exceptions=False)