From 4cffa3818c33737da63a4e4c5eab40d72426977b Mon Sep 17 00:00:00 2001 From: Bart Schuijt Date: Mon, 7 Sep 2026 22:02:27 +0200 Subject: [PATCH] refactor(adapters): implement polymorphic PipelineAdapter interface (tff#145) --- packages/tff-core/src/tff/core/adapter.py | 163 ++++++++ packages/tff-core/src/tff/core/autofix.py | 161 +++++--- packages/tff-core/src/tff/core/cli.py | 383 ++++++++--------- packages/tff-core/src/tff/core/docs.py | 79 ++-- packages/tff-core/src/tff/dataform/adapter.py | 83 ++++ packages/tff-core/src/tff/dataform/runner.py | 40 +- packages/tff-core/src/tff/dbt/adapter.py | 90 ++++ packages/tff-core/src/tff/dbt/runner.py | 35 +- packages/tff-core/src/tff/sqlmesh/adapter.py | 99 +++++ packages/tff-core/src/tff/sqlmesh/runner.py | 55 ++- packages/tff-core/tests/test_adapter.py | 384 ++++++++++++++++++ 11 files changed, 1234 insertions(+), 338 deletions(-) create mode 100644 packages/tff-core/src/tff/core/adapter.py create mode 100644 packages/tff-core/src/tff/dataform/adapter.py create mode 100644 packages/tff-core/src/tff/dbt/adapter.py create mode 100644 packages/tff-core/src/tff/sqlmesh/adapter.py create mode 100644 packages/tff-core/tests/test_adapter.py diff --git a/packages/tff-core/src/tff/core/adapter.py b/packages/tff-core/src/tff/core/adapter.py new file mode 100644 index 0000000..cf0d9d5 --- /dev/null +++ b/packages/tff-core/src/tff/core/adapter.py @@ -0,0 +1,163 @@ +"""Abstract PipelineAdapter interface and adapter registry.""" + +from __future__ import annotations + +import importlib +from abc import ABC, abstractmethod +from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from tff.core.config import FitnessFunctionsConfig + from tff.core.model import ModelRepresentation + from tff.core.report import LintFinding + + +class PipelineAdapter(ABC): + """Abstract interface for transformation pipeline engine adapters.""" + + @property + @abstractmethod + def provider_name(self) -> str: + """The identifier string for this provider, e.g. 'dbt', 'sqlmesh', 'dataform'.""" + raise NotImplementedError + + @abstractmethod + def is_applicable(self, project_root: Path) -> bool: + """Return True if project_root contains this adapter's configuration.""" + raise NotImplementedError + + @abstractmethod + def load_models( + self, + project_root: Path, + dialect: str | None = None, + manifest_path: str | Path | None = None, + ) -> dict[str, ModelRepresentation]: + """Load and map models from the pipeline project into ModelRepresentation objects.""" + raise NotImplementedError + + @abstractmethod + def run_checks( + self, + project_root: Path, + config: FitnessFunctionsConfig, + checks: list[str] | None = None, + dialect: str | None = None, + manifest_path: str | Path | None = None, + models: dict[str, ModelRepresentation] | None = None, + ) -> tuple[list[LintFinding], int, list[str]]: + """Run all enabled fitness functions and linter checks.""" + raise NotImplementedError + + def apply_metadata_fix( + self, + project_root: Path, + abs_path: Path, + model_name: str, + missing_owner: bool, + missing_description: bool, + ) -> str | None: + """Apply metadata fixes (owner, description) to model definition file if supported.""" + return None + + def get_diagnostic_files(self, project_root: Path) -> list[tuple[str, str]]: + """Return list of (file_label, display_status_str) for diagnostics.""" + return [] + + +ADAPTER_CLASSES: dict[str, tuple[str, str]] = { + "dbt": ("tff.dbt.adapter", "DBTAdapter"), + "sqlmesh": ("tff.sqlmesh.adapter", "SQLMeshAdapter"), + "dataform": ("tff.dataform.adapter", "DataformAdapter"), +} + + +def get_adapter(provider: str) -> PipelineAdapter: + """Load and return the adapter instance for the specified provider.""" + if provider not in ADAPTER_CLASSES: + raise ValueError(f"Unknown provider: {provider}") + + module_name, class_name = ADAPTER_CLASSES[provider] + + if provider == "dbt": + try: + mod = importlib.import_module(module_name) + except ImportError as e: + raise ImportError( + "dbt project detected, but tff is not installed with dbt support.\n" + 'Please install it using: pip install "tff-core[dbt]" or uv add "tff-core[dbt]"' + ) from e + elif provider == "sqlmesh": + try: + mod = importlib.import_module(module_name) + except ImportError as e: + raise ImportError( + "SQLMesh project detected, but tff is not installed with sqlmesh support.\n" + 'Please install it using: pip install "tff-core[sqlmesh]" or uv add "tff-core[sqlmesh]"' + ) from e + elif provider == "dataform": + try: + mod = importlib.import_module(module_name) + except ImportError as e: + raise ImportError( + "Dataform project detected, but tff is not installed with dataform support.\n" + 'Please install it using: pip install "tff-core[dataform]" or uv add "tff-core[dataform]"' + ) from e + else: + mod = importlib.import_module(module_name) + + adapter_cls: type[PipelineAdapter] = getattr(mod, class_name) + return adapter_cls() + + +def detect_provider(project_root: Path) -> str: + """Detect whether a project is dbt, SQLMesh, or Dataform.""" + # Check for dbt signature file + is_dbt = (project_root / "dbt_project.yml").exists() + + # Check for SQLMesh signature files + is_sqlmesh = ( + (project_root / ".sqlmesh").exists() + or (project_root / "config.py").exists() + or (project_root / "config.yaml").exists() + or (project_root / "config.yml").exists() + ) + + # Check for Dataform signature files + is_dataform = (project_root / "workflow_settings.yaml").exists() or ( + project_root / "dataform.json" + ).exists() + + detected = [ + p + for p, found in [ + ("dbt", is_dbt), + ("sqlmesh", is_sqlmesh), + ("dataform", is_dataform), + ] + if found + ] + + if is_dbt and is_sqlmesh and not is_dataform: + raise ValueError( + "Both dbt and SQLMesh configuration files were detected in the project root.\n" + "Please specify the provider explicitly using the --provider option (e.g. '--provider dbt' or '--provider sqlmesh')." + ) + if len(detected) > 1: + names = ", ".join(detected) + raise ValueError( + f"Multiple pipeline configuration files were detected in the project root ({names}).\n" + f"Please specify the provider explicitly using the --provider option (e.g. '--provider dbt', '--provider sqlmesh', or '--provider dataform')." + ) + if is_dbt: + return "dbt" + if is_sqlmesh: + return "sqlmesh" + if is_dataform: + return "dataform" + + raise ValueError( + "Could not detect project type (neither dbt_project.yml, SQLMesh config, nor Dataform config was found).\n" + "Please run this command from your project root, or specify the provider explicitly using the --provider option." + ) diff --git a/packages/tff-core/src/tff/core/autofix.py b/packages/tff-core/src/tff/core/autofix.py index 9f8758e..960e28f 100644 --- a/packages/tff-core/src/tff/core/autofix.py +++ b/packages/tff-core/src/tff/core/autofix.py @@ -11,6 +11,7 @@ from sqlglot import exp if TYPE_CHECKING: + from tff.core.adapter import PipelineAdapter from tff.core.model import ModelRepresentation from tff.core.report import LintFinding @@ -23,7 +24,7 @@ def parse_model_block_args(block_text: str) -> list[tuple[str, str]]: current = [] in_quotes = None depth = 0 - + # Split by top-level commas for char in block_text: if char in ("'", '"'): @@ -45,7 +46,7 @@ def parse_model_block_args(block_text: str) -> list[tuple[str, str]]: current.append(char) if current: pairs.append("".join(current).strip()) - + parsed_pairs = [] for pair in pairs: if not pair: @@ -61,6 +62,7 @@ def parse_model_block_args(block_text: str) -> list[tuple[str, str]]: def fix_positional_clauses(sql: str, dialect: str) -> str: """Rewrite positional GROUP BY and ORDER BY integers to explicit columns.""" from sqlglot.dialects.dialect import Dialect + resolved_dialect = None if dialect: try: @@ -70,11 +72,15 @@ def fix_positional_clauses(sql: str, dialect: str) -> str: pass # 1. Extract the SQLMesh MODEL or Dataform config block if present - model_block_match = re.match(r"^\s*(MODEL\s*\(.*?\)\s*;)", sql, flags=re.DOTALL | re.IGNORECASE) - config_match = re.search(r"^\s*config\s*\{", sql, flags=re.MULTILINE | re.IGNORECASE) + model_block_match = re.match( + r"^\s*(MODEL\s*\(.*?\)\s*;)", sql, flags=re.DOTALL | re.IGNORECASE + ) + config_match = re.search( + r"^\s*config\s*\{", sql, flags=re.MULTILINE | re.IGNORECASE + ) if model_block_match: model_block = model_block_match.group(1) - query_part = sql[model_block_match.end():] + query_part = sql[model_block_match.end() :] elif config_match: brace_start = config_match.end() - 1 depth = 0 @@ -105,7 +111,7 @@ def fix_positional_clauses(sql: str, dialect: str) -> str: else: model_block = "" query_part = sql - + # 2. Extract macros/Jinja to placeholders to prevent parsing errors patterns = [ r"\{#.*?#\}", @@ -117,27 +123,27 @@ def fix_positional_clauses(sql: str, dialect: str) -> str: ] combined_pattern = re.compile("|".join(patterns), re.DOTALL) placeholders = {} - + def repl(match): idx = len(placeholders) ph = f"__TFF_MACRO_PH_{idx}__" placeholders[ph] = match.group(0) return ph - + temp_query = combined_pattern.sub(repl, query_part) - + # 3. Parse with sqlglot try: parsed = sqlglot.parse_one(temp_query, read=resolved_dialect) except Exception: # If parsing fails, we cannot auto-fix this file return sql - + # 4. AST modification modified = False for select in parsed.find_all(exp.Select): selects = select.selects - + group = select.args.get("group") if group: new_group_expressions = [] @@ -156,7 +162,7 @@ def repl(match): else: new_group_expressions.append(expr) group.set("expressions", new_group_expressions) - + order = select.args.get("order") if order: for ordered in order.expressions: @@ -169,35 +175,37 @@ def repl(match): ordered.set("this", exp.column(select_expr.alias)) else: ordered.set("this", select_expr.copy()) - + if not modified: return sql - + # 5. Format back and restore placeholders modified_query = parsed.sql(dialect=resolved_dialect) for ph, orig in placeholders.items(): modified_query = modified_query.replace(ph, orig) - + if model_block: return model_block + "\n\n" + modified_query return modified_query -def fix_sqlmesh_metadata(abs_path: Path, missing_owner: bool, missing_description: bool) -> str | None: +def fix_sqlmesh_metadata( + abs_path: Path, missing_owner: bool, missing_description: bool +) -> str | None: """Update metadata fields inside a SQLMesh MODEL block.""" try: sql = abs_path.read_text(encoding="utf-8") except Exception: return None - + model_block_match = re.search(r"MODEL\s*\((.*?)\)", sql, re.DOTALL | re.IGNORECASE) if not model_block_match: return None - + args_str = model_block_match.group(1) args = parse_model_block_args(args_str) keys = {k.lower() for k, v in args} - + modified = False if missing_owner and "owner" not in keys: args.append(("owner", "'TODO: Add owner'")) @@ -205,14 +213,14 @@ def fix_sqlmesh_metadata(abs_path: Path, missing_owner: bool, missing_descriptio if missing_description and "description" not in keys: args.append(("description", "'TODO: Add description'")) modified = True - + if not modified: return None - + formatted_args = [f"{k} {v}" for k, v in args] new_block = "MODEL (\n " + ",\n ".join(formatted_args) + "\n)" new_sql = sql.replace(model_block_match.group(0), new_block, 1) - + try: abs_path.write_text(new_sql, encoding="utf-8") return f"Added missing metadata to MODEL block in {abs_path.name}" @@ -220,44 +228,61 @@ def fix_sqlmesh_metadata(abs_path: Path, missing_owner: bool, missing_descriptio return f"Failed to write SQLMesh metadata for {abs_path.name}: {e}" -def fix_dbt_metadata(abs_path: Path, model_name: str, missing_owner: bool, missing_description: bool) -> str | None: +def fix_dbt_metadata( + abs_path: Path, model_name: str, missing_owner: bool, missing_description: bool +) -> str | None: """Scaffold or update metadata fields for a dbt model in its directory's schema file.""" # Find any existing .yml/.yaml files in the same directory - yaml_files = list(abs_path.parent.glob("*.yml")) + list(abs_path.parent.glob("*.yaml")) - + yaml_files = list(abs_path.parent.glob("*.yml")) + list( + abs_path.parent.glob("*.yaml") + ) + for yf in yaml_files: try: with open(yf, encoding="utf-8") as f: data = yaml.safe_load(f) except Exception: continue - - if not isinstance(data, dict) or "models" not in data or not isinstance(data["models"], list): + + if ( + not isinstance(data, dict) + or "models" not in data + or not isinstance(data["models"], list) + ): continue - + # Look for the model entry model_entry = None for m in data["models"]: if isinstance(m, dict) and m.get("name") == model_name: model_entry = m break - + if model_entry is not None: modified = False - if missing_description and ("description" not in model_entry or not model_entry["description"]): + if missing_description and ( + "description" not in model_entry or not model_entry["description"] + ): model_entry["description"] = "TODO: Add description" modified = True if missing_owner: - if "meta" not in model_entry or not isinstance(model_entry["meta"], dict): + if "meta" not in model_entry or not isinstance( + model_entry["meta"], dict + ): model_entry["meta"] = {} - if "owner" not in model_entry["meta"] or not model_entry["meta"]["owner"]: + if ( + "owner" not in model_entry["meta"] + or not model_entry["meta"]["owner"] + ): model_entry["meta"]["owner"] = "TODO: Add owner" modified = True - + if modified: try: with open(yf, "w", encoding="utf-8") as f: - yaml.safe_dump(data, f, default_flow_style=False, sort_keys=False) + yaml.safe_dump( + data, f, default_flow_style=False, sort_keys=False + ) return f"Updated metadata for model {model_name} in {yf.name}" except Exception as e: return f"Failed to write dbt metadata to {yf.name}: {e}" @@ -273,23 +298,23 @@ def fix_dbt_metadata(abs_path: Path, model_name: str, missing_owner: bool, missi data = yaml.safe_load(f) or {} except Exception: pass - + if not isinstance(data, dict): data = {} - + if "version" not in data: data["version"] = 2 if "models" not in data or not isinstance(data["models"], list): data["models"] = [] - + model_entry = {"name": model_name} if missing_description: model_entry["description"] = "TODO: Add description" if missing_owner: model_entry["meta"] = {"owner": "TODO: Add owner"} - + data["models"].append(model_entry) - + try: with open(schema_path, "w", encoding="utf-8") as f: yaml.safe_dump(data, f, default_flow_style=False, sort_keys=False) @@ -302,26 +327,35 @@ def fix_dbt_metadata(abs_path: Path, model_name: str, missing_owner: bool, missi def apply_autofixes( project_root: Path, - provider: str, + provider: str | PipelineAdapter, findings: list[LintFinding], - models: dict[str, ModelRepresentation] + models: dict[str, ModelRepresentation], ) -> list[str]: """Identify auto-fixable violations from findings and apply modifications to source files.""" + from tff.core.adapter import get_adapter + + if isinstance(provider, str): + adapter = get_adapter(provider) + else: + adapter = provider + # Group findings by file path grouped = defaultdict(list) for f in findings: if f.path: abs_path = (project_root / f.path).resolve() grouped[abs_path].append(f) - + applied_logs = [] - + for abs_path, file_findings in grouped.items(): if not abs_path.exists(): continue - + # 1. Fix positional group by / order by - pos_findings = [f for f in file_findings if f.check == "nopositionalgroupbyororderby"] + pos_findings = [ + f for f in file_findings if f.check == "nopositionalgroupbyororderby" + ] if pos_findings and abs_path.suffix in (".sql", ".sqlx"): # Lookup dialect from models dictionary dialect = "ansi" @@ -329,30 +363,37 @@ def apply_autofixes( if Path(model.path).resolve() == abs_path: dialect = model.dialect break - + try: sql = abs_path.read_text(encoding="utf-8") fixed_sql = fix_positional_clauses(sql, dialect) if fixed_sql != sql: abs_path.write_text(fixed_sql, encoding="utf-8") - applied_logs.append(f"Fixed positional GROUP BY/ORDER BY in {abs_path.name}") + applied_logs.append( + f"Fixed positional GROUP BY/ORDER BY in {abs_path.name}" + ) except Exception as e: - applied_logs.append(f"Failed to fix positional references in {abs_path.name}: {e}") - + applied_logs.append( + f"Failed to fix positional references in {abs_path.name}: {e}" + ) + # 2. Fix metadata issues (owner, description) missing_owner = any(f.check == "nomissingowner" for f in file_findings) - missing_description = any(f.check == "nomissingdescription" for f in file_findings) - + missing_description = any( + f.check == "nomissingdescription" for f in file_findings + ) + if missing_owner or missing_description: model_name = file_findings[0].model if model_name: - if provider == "sqlmesh": - log = fix_sqlmesh_metadata(abs_path, missing_owner, missing_description) - if log: - applied_logs.append(log) - elif provider == "dbt": - log = fix_dbt_metadata(abs_path, model_name, missing_owner, missing_description) - if log: - applied_logs.append(log) - + log = adapter.apply_metadata_fix( + project_root=project_root, + abs_path=abs_path, + model_name=model_name, + missing_owner=missing_owner, + missing_description=missing_description, + ) + if log: + applied_logs.append(log) + return applied_logs diff --git a/packages/tff-core/src/tff/core/cli.py b/packages/tff-core/src/tff/core/cli.py index ef253d4..d16bc46 100644 --- a/packages/tff-core/src/tff/core/cli.py +++ b/packages/tff-core/src/tff/core/cli.py @@ -8,12 +8,18 @@ import logging import sys from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any +from tff.core.adapter import PipelineAdapter, detect_provider, get_adapter from tff.core.config import load_fitness_config, resolve_project_path from tff.core.context import set_ff_config from tff.core.report import render_lint_report +if TYPE_CHECKING: + from tff.core.config import FitnessFunctionsConfig + from tff.core.model import ModelRepresentation + from tff.core.report import LintFinding + try: __version__ = importlib.metadata.version("tff-core") except Exception: @@ -22,47 +28,7 @@ def _detect_provider(project_root: Path) -> str: """Detect whether a project is dbt, SQLMesh, or Dataform.""" - # Check for dbt signature file - is_dbt = (project_root / "dbt_project.yml").exists() - - # Check for SQLMesh signature files - is_sqlmesh = ( - (project_root / ".sqlmesh").exists() - or (project_root / "config.py").exists() - or (project_root / "config.yaml").exists() - or (project_root / "config.yml").exists() - ) - - # Check for Dataform signature files - is_dataform = ( - (project_root / "workflow_settings.yaml").exists() - or (project_root / "dataform.json").exists() - ) - - detected = [p for p, found in [("dbt", is_dbt), ("sqlmesh", is_sqlmesh), ("dataform", is_dataform)] if found] - - if is_dbt and is_sqlmesh and not is_dataform: - raise ValueError( - "Both dbt and SQLMesh configuration files were detected in the project root.\n" - "Please specify the provider explicitly using the --provider option (e.g. '--provider dbt' or '--provider sqlmesh')." - ) - if len(detected) > 1: - names = ", ".join(detected) - raise ValueError( - f"Multiple pipeline configuration files were detected in the project root ({names}).\n" - "Please specify the provider explicitly using the --provider option (e.g. '--provider dbt', '--provider sqlmesh', or '--provider dataform')." - ) - if is_dbt: - return "dbt" - if is_sqlmesh: - return "sqlmesh" - if is_dataform: - return "dataform" - - raise ValueError( - "Could not detect project type (neither dbt_project.yml, SQLMesh config, nor Dataform config was found).\n" - "Please run this command from your project root, or specify the provider explicitly using the --provider option." - ) + return detect_provider(project_root) def _get_runner(provider: str) -> Any: @@ -95,6 +61,142 @@ def _get_runner(provider: str) -> Any: raise ValueError(f"Unknown provider: {provider}") +class _MockRunnerAdapter(PipelineAdapter): + """Fallback adapter wrapping a mock runner for backwards-compatibility with existing tests.""" + + def __init__(self, provider: str, mock_runner: Any) -> None: + self._provider = provider + self._runner = mock_runner + + @property + def provider_name(self) -> str: + return self._provider + + def is_applicable(self, project_root: Path) -> bool: + return True + + def load_models( + self, + project_root: Path, + dialect: str | None = None, + manifest_path: str | Path | None = None, + ) -> dict[str, Any]: + if self._provider == "dbt": + from tff.dbt.manifest import load_dbt_models + + return load_dbt_models(project_root, dialect=dialect) + elif self._provider == "dataform": + from tff.dataform.manifest import load_dataform_models + + return load_dataform_models( + project_root, manifest_path=manifest_path, dialect=dialect + ) + elif self._provider == "sqlmesh": + from sqlmesh.core.context import Context + from tff.sqlmesh.loader import FitnessLoader + from tff.sqlmesh.runner import map_sqlmesh_context_models + + context = Context(paths=[str(project_root)], loader=FitnessLoader) + return map_sqlmesh_context_models(context) + return {} + + def run_checks( + self, + project_root: Path, + config: FitnessFunctionsConfig, + checks: list[str] | None = None, + dialect: str | None = None, + manifest_path: str | Path | None = None, + models: dict[str, ModelRepresentation] | None = None, + ) -> tuple[list[LintFinding], int, list[str]]: + kwargs: dict[str, Any] = { + "project_root": project_root, + "config": config, + "checks": checks, + } + if self._provider == "dbt": + kwargs["dialect"] = dialect + if models is not None: + kwargs["models"] = models + elif self._provider == "dataform": + kwargs["dialect"] = dialect + kwargs["manifest_path"] = manifest_path + if models is not None: + kwargs["models"] = models + elif self._provider == "sqlmesh": + if models is not None: + kwargs["models"] = models + return self._runner.run_all_checks(**kwargs) + + def apply_metadata_fix( + self, + project_root: Path, + abs_path: Path, + model_name: str, + missing_owner: bool, + missing_description: bool, + ) -> str | None: + if self._provider == "dbt": + from tff.core.autofix import fix_dbt_metadata + + return fix_dbt_metadata( + abs_path=abs_path, + model_name=model_name, + missing_owner=missing_owner, + missing_description=missing_description, + ) + elif self._provider == "sqlmesh": + from tff.core.autofix import fix_sqlmesh_metadata + + return fix_sqlmesh_metadata( + abs_path=abs_path, + missing_owner=missing_owner, + missing_description=missing_description, + ) + return None + + def get_diagnostic_files(self, project_root: Path) -> list[tuple[str, str]]: + if self._provider == "dbt": + dbt_project = project_root / "dbt_project.yml" + manifest = project_root / "target" / "manifest.json" + dbt_project_status = ( + "[green]found[/green]" if dbt_project.exists() else "[red]missing[/red]" + ) + manifest_status = ( + "[green]found[/green]" if manifest.exists() else "[red]missing[/red]" + ) + return [ + ("dbt_project.yml", f"{dbt_project} ({dbt_project_status})"), + ("manifest.json", f"{manifest} ({manifest_status})"), + ] + elif self._provider == "sqlmesh": + config_py = project_root / "config.py" + settings_yaml = project_root / "settings.yaml" + config_py_status = ( + "[green]found[/green]" if config_py.exists() else "[red]missing[/red]" + ) + settings_yaml_status = ( + "[green]found[/green]" + if settings_yaml.exists() + else "[red]missing[/red]" + ) + return [ + ("config.py", f"{config_py} ({config_py_status})"), + ("settings.yaml", f"{settings_yaml} ({settings_yaml_status})"), + ] + return [] + + +def _get_adapter(provider: str) -> PipelineAdapter: + """Load and return the adapter instance for the specified provider, wrapping test mocks if present.""" + runner = _get_runner(provider) + from unittest.mock import Mock + + if isinstance(runner, Mock): + return _MockRunnerAdapter(provider, runner) + return get_adapter(provider) + + def _parse_checks(value: str | None) -> list[str] | None: if not value: return None @@ -535,74 +637,18 @@ def format_ver(pkg: str) -> str: ) console.print(ver_table) + try: + adapter = _get_adapter(provider) + except (ImportError, ValueError) as e: + console.print(f"[red]Error loading adapter: {e}[/red]") + return 1 + prov_table = Table(show_header=False, box=None, padding=(0, 2, 0, 0)) prov_table.add_column() prov_table.add_column() - if provider == "dbt": - dbt_project = project_root / "dbt_project.yml" - manifest = project_root / "target" / "manifest.json" - dbt_project_status = ( - "[green]found[/green]" if dbt_project.exists() else "[red]missing[/red]" - ) - manifest_status = ( - "[green]found[/green]" if manifest.exists() else "[red]missing[/red]" - ) - prov_table.add_row( - " [bold]dbt_project.yml[/bold]", - f"{dbt_project} ({dbt_project_status})", - ) - prov_table.add_row( - " [bold]manifest.json[/bold]", - f"{manifest} ({manifest_status})", - ) - elif provider == "sqlmesh": - config_py = project_root / "config.py" - settings_yaml = project_root / "settings.yaml" - config_py_status = ( - "[green]found[/green]" if config_py.exists() else "[red]missing[/red]" - ) - settings_yaml_status = ( - "[green]found[/green]" - if settings_yaml.exists() - else "[red]missing[/red]" - ) - prov_table.add_row( - " [bold]config.py[/bold]", - f"{config_py} ({config_py_status})", - ) - prov_table.add_row( - " [bold]settings.yaml[/bold]", - f"{settings_yaml} ({settings_yaml_status})", - ) - elif provider == "dataform": - ws_yaml = project_root / "workflow_settings.yaml" - df_json = project_root / "dataform.json" - if ws_yaml.exists(): - prov_table.add_row( - " [bold]workflow_settings.yaml[/bold]", - f"{ws_yaml} ([green]found[/green])", - ) - elif df_json.exists(): - prov_table.add_row( - " [bold]dataform.json[/bold]", - f"{df_json} ([green]found[/green])", - ) - else: - prov_table.add_row( - " [bold]workflow_settings.yaml[/bold]", - "[red]missing[/red]", - ) - - from tff.dataform.manifest import _find_manifest_file - - found_manifest = _find_manifest_file(project_root) - m_status = ( - f"[green]{found_manifest.name}[/green]" - if found_manifest - else "[dim]not found (will compile via CLI or parse .sqlx)[/dim]" - ) - prov_table.add_row(" [bold]compilation manifest[/bold]", m_status) + for label, val in adapter.get_diagnostic_files(project_root): + prov_table.add_row(f" [bold]{label}[/bold]", val) if prov_table.row_count > 0: console.print("\n[bold cyan]● Provider Files[/bold cyan]") console.print(prov_table) @@ -743,6 +789,7 @@ def format_ver(pkg: str) -> str: return 1 from tff.core.docs import generate_docs_dashboard + docs_kwargs: dict[str, Any] = { "project_root": project_root, "output_path": args.output, @@ -773,9 +820,9 @@ def format_ver(pkg: str) -> str: print(f"Error: {e}", file=sys.stderr) return 1 - # 2. Get runner (checks adapter availability) + # 2. Get adapter (checks adapter availability) try: - runner_module = _get_runner(provider) + adapter = _get_adapter(provider) except (ImportError, ValueError) as e: print(f"Error: {e}", file=sys.stderr) return 1 @@ -797,106 +844,64 @@ def format_ver(pkg: str) -> str: checks = None # Always run all checks for health report # 4. Run checks + if provider == "sqlmesh" and args.dialect is not None: + print( + "Warning: --dialect is ignored for SQLMesh projects (dialects are defined directly on models).", + file=sys.stderr, + ) + manifest_path = getattr(args, "manifest", None) try: - if provider == "dbt": - findings, models_checked, executed_checks = ( - runner_module.run_all_checks( - project_root=project_root, - config=config, - checks=checks, - dialect=args.dialect, - ) - ) - elif provider == "dataform": - findings, models_checked, executed_checks = ( - runner_module.run_all_checks( - project_root=project_root, - config=config, - checks=checks, - dialect=args.dialect, - manifest_path=manifest_path, - ) - ) - else: - if args.dialect is not None: - print( - "Warning: --dialect is ignored for SQLMesh projects (dialects are defined directly on models).", - file=sys.stderr, - ) - findings, models_checked, executed_checks = ( - runner_module.run_all_checks( - project_root=project_root, - config=config, - checks=checks, - ) - ) + findings, models_checked, executed_checks = adapter.run_checks( + project_root=project_root, + config=config, + checks=checks, + dialect=args.dialect, + manifest_path=manifest_path, + ) except Exception as e: print(f"Error executing checks: {e}", file=sys.stderr) return 1 # Apply auto-fixes if --fix is set if args.command == "lint" and getattr(args, "fix", False) and findings: - models = {} try: - if provider == "dbt": - from tff.dbt.manifest import load_dbt_models - models = load_dbt_models(project_root, dialect=args.dialect) - elif provider == "dataform": - from tff.dataform.manifest import load_dataform_models - models = load_dataform_models(project_root, manifest_path=manifest_path, dialect=args.dialect) - else: - from sqlmesh.core.context import Context - from tff.sqlmesh.loader import FitnessLoader - from tff.sqlmesh.runner import map_sqlmesh_context_models - context = Context( - paths=[str(project_root)], - loader=FitnessLoader, - ) - models = map_sqlmesh_context_models(context) + models = adapter.load_models( + project_root=project_root, + dialect=args.dialect, + manifest_path=manifest_path, + ) except Exception as e: - print(f"Warning: Could not load models for autofix: {e}", file=sys.stderr) + models = {} + print( + f"Warning: Could not load models for autofix: {e}", file=sys.stderr + ) if models: from tff.core.autofix import apply_autofixes - fix_logs = apply_autofixes(project_root, provider, findings, models) + + fix_logs = apply_autofixes(project_root, adapter, findings, models) if fix_logs: if not args.json: from rich.console import Console + console = Console(stderr=True) for log in fix_logs: console.print(f"[green]✓[/green] {log}") # Re-run checks to get the final state of the files try: - if provider == "dbt": - findings, models_checked, executed_checks = ( - runner_module.run_all_checks( - project_root=project_root, - config=config, - checks=checks, - dialect=args.dialect, - ) - ) - elif provider == "dataform": - findings, models_checked, executed_checks = ( - runner_module.run_all_checks( - project_root=project_root, - config=config, - checks=checks, - dialect=args.dialect, - manifest_path=manifest_path, - ) - ) - else: - findings, models_checked, executed_checks = ( - runner_module.run_all_checks( - project_root=project_root, - config=config, - checks=checks, - ) - ) + findings, models_checked, executed_checks = adapter.run_checks( + project_root=project_root, + config=config, + checks=checks, + dialect=args.dialect, + manifest_path=manifest_path, + ) except Exception as e: - print(f"Error executing checks after autofix: {e}", file=sys.stderr) + print( + f"Error executing checks after autofix: {e}", + file=sys.stderr, + ) return 1 if args.command == "lint": @@ -951,7 +956,11 @@ def format_ver(pkg: str) -> str: scoped_models_count=scoped_models_count, ) - effective_models_checked = scoped_models_count if scoped_models_count is not None else models_checked + effective_models_checked = ( + scoped_models_count + if scoped_models_count is not None + else models_checked + ) json_data = get_health_json_data(scores, effective_models_checked) save_log(project_root, "health", json_data) diff --git a/packages/tff-core/src/tff/core/docs.py b/packages/tff-core/src/tff/core/docs.py index 5dc5c93..5341dd9 100644 --- a/packages/tff-core/src/tff/core/docs.py +++ b/packages/tff-core/src/tff/core/docs.py @@ -27,47 +27,26 @@ def generate_docs_dashboard( config = load_fitness_config(project_root, config_path=config_path) set_ff_config(config) - # 2. Get runner - from tff.core.cli import _get_runner - runner_module = _get_runner(provider) + # 2. Get adapter + from tff.core.cli import _get_adapter + + adapter = _get_adapter(provider) # 3. Load models mapping - models = {} - if provider == "dbt": - from tff.dbt.manifest import load_dbt_models - models = load_dbt_models(project_root, dialect=dialect) - elif provider == "dataform": - from tff.dataform.manifest import load_dataform_models - models = load_dataform_models(project_root, manifest_path=manifest_path, dialect=dialect) - else: - from sqlmesh.core.context import Context - from tff.sqlmesh.loader import FitnessLoader - from tff.sqlmesh.runner import map_sqlmesh_context_models - context = Context( - paths=[str(project_root)], - loader=FitnessLoader, - ) - models = map_sqlmesh_context_models(context) - - # 4. Run all checks - if provider == "dbt": - findings, models_checked, executed_checks = runner_module.run_all_checks( - project_root=project_root, - config=config, - dialect=dialect, - ) - elif provider == "dataform": - findings, models_checked, executed_checks = runner_module.run_all_checks( - project_root=project_root, - config=config, - dialect=dialect, - manifest_path=manifest_path, - ) - else: - findings, models_checked, executed_checks = runner_module.run_all_checks( - project_root=project_root, - config=config, - ) + models = adapter.load_models( + project_root=project_root, + dialect=dialect, + manifest_path=manifest_path, + ) + + # 4. Run all checks reusing preloaded models + findings, models_checked, executed_checks = adapter.run_checks( + project_root=project_root, + config=config, + dialect=dialect, + manifest_path=manifest_path, + models=models, + ) # 5. Calculate scores and save health log scores = calculate_health_scores(findings, models_checked, config, provider) @@ -77,15 +56,19 @@ def generate_docs_dashboard( # 6. Collect history (60 days) history = collect_stats(project_root, days=60) if not history: - history = [{ - "date": date.today().isoformat(), - "health_score": scores["overall_score"], - "errors_count": len([f for f in findings if f.severity == "error"]), - "warnings_count": len([f for f in findings if f.severity == "warning"]), - }] + history = [ + { + "date": date.today().isoformat(), + "health_score": scores["overall_score"], + "errors_count": len([f for f in findings if f.severity == "error"]), + "warnings_count": len([f for f in findings if f.severity == "warning"]), + } + ] # 7. Prep data for the HTML template - rel_to_model_id = {model_path_relative(m): m_id for m_id, m in models.items() if m.path} + rel_to_model_id = { + model_path_relative(m): m_id for m_id, m in models.items() if m.path + } name_to_model_id = {m.name: m_id for m_id, m in models.items()} model_findings_serialized: dict[str, list[dict[str, Any]]] = {} @@ -150,7 +133,9 @@ def generate_docs_dashboard( } embedded_json = json.dumps(data_to_embed, indent=2) - html_content = HTML_TEMPLATE.replace("", f"const TFF_DATA = {embedded_json};") + html_content = HTML_TEMPLATE.replace( + "", f"const TFF_DATA = {embedded_json};" + ) if output_path is None: output_path = project_root / "tff_report.html" diff --git a/packages/tff-core/src/tff/dataform/adapter.py b/packages/tff-core/src/tff/dataform/adapter.py new file mode 100644 index 0000000..8707d79 --- /dev/null +++ b/packages/tff-core/src/tff/dataform/adapter.py @@ -0,0 +1,83 @@ +"""Dataform adapter implementation for TFF.""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +from tff.core.adapter import PipelineAdapter + +if TYPE_CHECKING: + from tff.core.config import FitnessFunctionsConfig + from tff.core.model import ModelRepresentation + from tff.core.report import LintFinding + + +class DataformAdapter(PipelineAdapter): + """Pipeline adapter for Dataform projects.""" + + @property + def provider_name(self) -> str: + return "dataform" + + def is_applicable(self, project_root: Path) -> bool: + return (project_root / "workflow_settings.yaml").exists() or ( + project_root / "dataform.json" + ).exists() + + def load_models( + self, + project_root: Path, + dialect: str | None = None, + manifest_path: str | Path | None = None, + ) -> dict[str, ModelRepresentation]: + from tff.dataform.manifest import load_dataform_models + + return load_dataform_models( + project_root=project_root, + manifest_path=manifest_path, + dialect=dialect, + ) + + def run_checks( + self, + project_root: Path, + config: FitnessFunctionsConfig, + checks: list[str] | None = None, + dialect: str | None = None, + manifest_path: str | Path | None = None, + models: dict[str, ModelRepresentation] | None = None, + ) -> tuple[list[LintFinding], int, list[str]]: + from tff.dataform.runner import run_all_checks + + return run_all_checks( + project_root=project_root, + config=config, + checks=checks, + dialect=dialect, + manifest_path=manifest_path, + models=models, + ) + + def get_diagnostic_files(self, project_root: Path) -> list[tuple[str, str]]: + ws_yaml = project_root / "workflow_settings.yaml" + df_json = project_root / "dataform.json" + from tff.dataform.manifest import _find_manifest_file + + found_manifest = _find_manifest_file(project_root) + m_status = ( + f"[green]{found_manifest.name}[/green]" + if found_manifest + else "[dim]not found (will compile via CLI or parse .sqlx)[/dim]" + ) + + rows: list[tuple[str, str]] = [] + if ws_yaml.exists(): + rows.append(("workflow_settings.yaml", f"{ws_yaml} ([green]found[/green])")) + elif df_json.exists(): + rows.append(("dataform.json", f"{df_json} ([green]found[/green])")) + else: + rows.append(("workflow_settings.yaml", "[red]missing[/red]")) + + rows.append(("compilation manifest", m_status)) + return rows diff --git a/packages/tff-core/src/tff/dataform/runner.py b/packages/tff-core/src/tff/dataform/runner.py index aabb17b..077b5d2 100644 --- a/packages/tff-core/src/tff/dataform/runner.py +++ b/packages/tff-core/src/tff/dataform/runner.py @@ -23,17 +23,29 @@ logger = logging.getLogger(__name__) CHECK_COLLECTORS = { - "layer_integrity": lambda models, cfg: collect_layer_integrity_findings(models, cfg), - "custom_exclusions": lambda models, cfg: collect_custom_exclusion_findings(models, cfg), + "layer_integrity": lambda models, cfg: collect_layer_integrity_findings( + models, cfg + ), + "custom_exclusions": lambda models, cfg: collect_custom_exclusion_findings( + models, cfg + ), "schema_contracts": lambda _models, cfg: collect_schema_contract_findings(cfg), - "dependency_graph": lambda models, cfg: collect_dependency_graph_findings(models, cfg), - "materialization_depth": lambda models, cfg: collect_materialization_depth_findings(models, cfg), + "dependency_graph": lambda models, cfg: collect_dependency_graph_findings( + models, cfg + ), + "materialization_depth": lambda models, cfg: collect_materialization_depth_findings( + models, cfg + ), "duplicate_ctes": lambda models, cfg: collect_duplicate_cte_findings(models, cfg), - "connascence_of_value": lambda models, cfg: collect_connascence_of_value_findings(models, cfg), + "connascence_of_value": lambda models, cfg: collect_connascence_of_value_findings( + models, cfg + ), } -def collect_dataform_rules_findings(models: dict[str, ModelRepresentation]) -> list[LintFinding]: +def collect_dataform_rules_findings( + models: dict[str, ModelRepresentation], +) -> list[LintFinding]: findings = [] rules = [rule_cls() for rule_cls in ALL_RULES] @@ -74,20 +86,22 @@ def run_all_checks( checks: list[str] | None = None, dialect: str | None = None, manifest_path: str | Path | None = None, + models: dict[str, ModelRepresentation] | None = None, ) -> tuple[list[LintFinding], int, list[str]]: project_root = project_root or Path.cwd() if config is None: config = load_fitness_config(project_root) set_ff_config(config) - # Load Dataform models - models = load_dataform_models(project_root, manifest_path=manifest_path, dialect=dialect) + # Load Dataform models if not already provided + if models is None: + models = load_dataform_models( + project_root, manifest_path=manifest_path, dialect=dialect + ) if checks is None: selected = ["rules"] + [ - name - for name in CHECK_COLLECTORS - if _check_enabled(config, name) + name for name in CHECK_COLLECTORS if _check_enabled(config, name) ] else: selected = checks @@ -103,6 +117,8 @@ def run_all_checks( findings.extend(collector(models, config)) # Count of non-external, non-symbolic models checked - models_checked = sum(1 for m in models.values() if not m.is_external and not m.is_symbolic) + models_checked = sum( + 1 for m in models.values() if not m.is_external and not m.is_symbolic + ) return findings, models_checked, selected diff --git a/packages/tff-core/src/tff/dbt/adapter.py b/packages/tff-core/src/tff/dbt/adapter.py new file mode 100644 index 0000000..efd7b11 --- /dev/null +++ b/packages/tff-core/src/tff/dbt/adapter.py @@ -0,0 +1,90 @@ +"""dbt adapter implementation for TFF.""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +from tff.core.adapter import PipelineAdapter + +if TYPE_CHECKING: + from tff.core.config import FitnessFunctionsConfig + from tff.core.model import ModelRepresentation + from tff.core.report import LintFinding + + +class DBTAdapter(PipelineAdapter): + """Pipeline adapter for dbt projects.""" + + @property + def provider_name(self) -> str: + return "dbt" + + def is_applicable(self, project_root: Path) -> bool: + return (project_root / "dbt_project.yml").exists() + + def load_models( + self, + project_root: Path, + dialect: str | None = None, + manifest_path: str | Path | None = None, + ) -> dict[str, ModelRepresentation]: + from tff.dbt.manifest import load_dbt_models + + return load_dbt_models(project_root, dialect=dialect) + + def run_checks( + self, + project_root: Path, + config: FitnessFunctionsConfig, + checks: list[str] | None = None, + dialect: str | None = None, + manifest_path: str | Path | None = None, + models: dict[str, ModelRepresentation] | None = None, + ) -> tuple[list[LintFinding], int, list[str]]: + from tff.dbt.runner import run_all_checks + + return run_all_checks( + project_root=project_root, + config=config, + checks=checks, + dialect=dialect, + models=models, + ) + + def apply_metadata_fix( + self, + project_root: Path, + abs_path: Path, + model_name: str, + missing_owner: bool, + missing_description: bool, + ) -> str | None: + from tff.core.autofix import fix_dbt_metadata + + return fix_dbt_metadata( + abs_path=abs_path, + model_name=model_name, + missing_owner=missing_owner, + missing_description=missing_description, + ) + + def get_diagnostic_files(self, project_root: Path) -> list[tuple[str, str]]: + dbt_project = project_root / "dbt_project.yml" + manifest = project_root / "target" / "manifest.json" + dbt_project_status = ( + "[green]found[/green]" if dbt_project.exists() else "[red]missing[/red]" + ) + manifest_status = ( + "[green]found[/green]" if manifest.exists() else "[red]missing[/red]" + ) + return [ + ( + "dbt_project.yml", + f"{dbt_project} ({dbt_project_status})", + ), + ( + "manifest.json", + f"{manifest} ({manifest_status})", + ), + ] diff --git a/packages/tff-core/src/tff/dbt/runner.py b/packages/tff-core/src/tff/dbt/runner.py index a1ddea7..bc774d3 100644 --- a/packages/tff-core/src/tff/dbt/runner.py +++ b/packages/tff-core/src/tff/dbt/runner.py @@ -23,18 +23,29 @@ logger = logging.getLogger(__name__) CHECK_COLLECTORS = { - "layer_integrity": lambda models, cfg: collect_layer_integrity_findings(models, cfg), - "custom_exclusions": lambda models, cfg: collect_custom_exclusion_findings(models, cfg), + "layer_integrity": lambda models, cfg: collect_layer_integrity_findings( + models, cfg + ), + "custom_exclusions": lambda models, cfg: collect_custom_exclusion_findings( + models, cfg + ), "schema_contracts": lambda _models, cfg: collect_schema_contract_findings(cfg), - "dependency_graph": lambda models, cfg: collect_dependency_graph_findings(models, cfg), - "materialization_depth": lambda models, cfg: collect_materialization_depth_findings(models, cfg), + "dependency_graph": lambda models, cfg: collect_dependency_graph_findings( + models, cfg + ), + "materialization_depth": lambda models, cfg: collect_materialization_depth_findings( + models, cfg + ), "duplicate_ctes": lambda models, cfg: collect_duplicate_cte_findings(models, cfg), - "connascence_of_value": lambda models, cfg: collect_connascence_of_value_findings(models, cfg), + "connascence_of_value": lambda models, cfg: collect_connascence_of_value_findings( + models, cfg + ), } - -def collect_dbt_rules_findings(models: dict[str, ModelRepresentation]) -> list[LintFinding]: +def collect_dbt_rules_findings( + models: dict[str, ModelRepresentation], +) -> list[LintFinding]: findings = [] rules = [rule_cls() for rule_cls in ALL_RULES] @@ -75,20 +86,20 @@ def run_all_checks( config: FitnessFunctionsConfig | None = None, checks: list[str] | None = None, dialect: str | None = None, + models: dict[str, ModelRepresentation] | None = None, ) -> tuple[list[LintFinding], int, list[str]]: project_root = project_root or Path.cwd() if config is None: config = load_fitness_config(project_root) set_ff_config(config) - # Parse and load manifest.json - models = load_dbt_models(project_root, dialect=dialect) + # Parse and load manifest.json if models not already provided + if models is None: + models = load_dbt_models(project_root, dialect=dialect) if checks is None: selected = ["rules"] + [ - name - for name in CHECK_COLLECTORS - if _check_enabled(config, name) + name for name in CHECK_COLLECTORS if _check_enabled(config, name) ] else: selected = checks diff --git a/packages/tff-core/src/tff/sqlmesh/adapter.py b/packages/tff-core/src/tff/sqlmesh/adapter.py new file mode 100644 index 0000000..e1f1af4 --- /dev/null +++ b/packages/tff-core/src/tff/sqlmesh/adapter.py @@ -0,0 +1,99 @@ +"""SQLMesh adapter implementation for TFF.""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +from tff.core.adapter import PipelineAdapter + +if TYPE_CHECKING: + from tff.core.config import FitnessFunctionsConfig + from tff.core.model import ModelRepresentation + from tff.core.report import LintFinding + + +class SQLMeshAdapter(PipelineAdapter): + """Pipeline adapter for SQLMesh projects.""" + + @property + def provider_name(self) -> str: + return "sqlmesh" + + def is_applicable(self, project_root: Path) -> bool: + return ( + (project_root / ".sqlmesh").exists() + or (project_root / "config.py").exists() + or (project_root / "config.yaml").exists() + or (project_root / "config.yml").exists() + ) + + def load_models( + self, + project_root: Path, + dialect: str | None = None, + manifest_path: str | Path | None = None, + ) -> dict[str, ModelRepresentation]: + from sqlmesh.core.context import Context + from tff.sqlmesh.loader import FitnessLoader + from tff.sqlmesh.runner import map_sqlmesh_context_models + + context = Context( + paths=[str(project_root)], + loader=FitnessLoader, + ) + return map_sqlmesh_context_models(context) + + def run_checks( + self, + project_root: Path, + config: FitnessFunctionsConfig, + checks: list[str] | None = None, + dialect: str | None = None, + manifest_path: str | Path | None = None, + models: dict[str, ModelRepresentation] | None = None, + ) -> tuple[list[LintFinding], int, list[str]]: + from tff.sqlmesh.runner import run_all_checks + + return run_all_checks( + project_root=project_root, + config=config, + checks=checks, + models=models, + ) + + def apply_metadata_fix( + self, + project_root: Path, + abs_path: Path, + model_name: str, + missing_owner: bool, + missing_description: bool, + ) -> str | None: + from tff.core.autofix import fix_sqlmesh_metadata + + return fix_sqlmesh_metadata( + abs_path=abs_path, + missing_owner=missing_owner, + missing_description=missing_description, + ) + + def get_diagnostic_files(self, project_root: Path) -> list[tuple[str, str]]: + config_py = project_root / "config.py" + settings_yaml = project_root / "settings.yaml" + config_py_status = ( + "[green]found[/green]" if config_py.exists() else "[red]missing[/red]" + ) + settings_yaml_status = ( + "[green]found[/green]" if settings_yaml.exists() else "[red]missing[/red]" + ) + return [ + ( + "config.py", + f"{config_py} ({config_py_status})", + ), + ( + "settings.yaml", + f"{settings_yaml} ({settings_yaml_status})", + ), + ] diff --git a/packages/tff-core/src/tff/sqlmesh/runner.py b/packages/tff-core/src/tff/sqlmesh/runner.py index c8ace94..409b11d 100644 --- a/packages/tff-core/src/tff/sqlmesh/runner.py +++ b/packages/tff-core/src/tff/sqlmesh/runner.py @@ -25,17 +25,26 @@ logger = logging.getLogger(__name__) CHECK_COLLECTORS = { - "layer_integrity": lambda models, cfg: collect_layer_integrity_findings(models, cfg), - "custom_exclusions": lambda models, cfg: collect_custom_exclusion_findings(models, cfg), + "layer_integrity": lambda models, cfg: collect_layer_integrity_findings( + models, cfg + ), + "custom_exclusions": lambda models, cfg: collect_custom_exclusion_findings( + models, cfg + ), "schema_contracts": lambda _models, cfg: collect_schema_contract_findings(cfg), - "dependency_graph": lambda models, cfg: collect_dependency_graph_findings(models, cfg), - "materialization_depth": lambda models, cfg: collect_materialization_depth_findings(models, cfg), + "dependency_graph": lambda models, cfg: collect_dependency_graph_findings( + models, cfg + ), + "materialization_depth": lambda models, cfg: collect_materialization_depth_findings( + models, cfg + ), "duplicate_ctes": lambda models, cfg: collect_duplicate_cte_findings(models, cfg), - "connascence_of_value": lambda models, cfg: collect_connascence_of_value_findings(models, cfg), + "connascence_of_value": lambda models, cfg: collect_connascence_of_value_findings( + models, cfg + ), } - class _SilentLinterConsole: def show_linter_violations(self, *args, **kwargs) -> None: return None @@ -84,9 +93,7 @@ def collect_sqlmesh_findings(context: Context) -> list[LintFinding]: def count_models_checked(context: Context) -> int: - return sum( - 1 for model in context.models.values() if not model.kind.is_symbolic - ) + return sum(1 for model in context.models.values() if not model.kind.is_symbolic) def _check_enabled(config: FitnessFunctionsConfig, check_name: str) -> bool: @@ -106,32 +113,34 @@ def run_all_checks( context: Context | None = None, config: FitnessFunctionsConfig | None = None, checks: list[str] | None = None, + models: dict[str, ModelRepresentation] | None = None, ) -> tuple[list[LintFinding], int, list[str]]: project_root = project_root or Path.cwd() if config is None: config = load_fitness_config(project_root) set_ff_config(config) - context = context or Context( - paths=[str(project_root)], - loader=FitnessLoader, - ) - if checks is None: selected = ["sqlmesh"] + [ - name - for name in CHECK_COLLECTORS - if _check_enabled(config, name) + name for name in CHECK_COLLECTORS if _check_enabled(config, name) ] else: selected = checks findings: list[LintFinding] = [] - if "sqlmesh" in selected: + if "sqlmesh" in selected or models is None: + context = context or Context( + paths=[str(project_root)], + loader=FitnessLoader, + ) + + if "sqlmesh" in selected and context is not None: findings.extend(collect_sqlmesh_findings(context)) - mapped_models = map_sqlmesh_context_models(context) + mapped_models = ( + models if models is not None else map_sqlmesh_context_models(context) + ) for check_name, collector in CHECK_COLLECTORS.items(): if check_name not in selected: @@ -140,4 +149,10 @@ def run_all_checks( continue findings.extend(collector(mapped_models, config)) - return findings, count_models_checked(context), selected + checked_count = ( + count_models_checked(context) + if context is not None + else sum(1 for m in mapped_models.values() if not m.is_symbolic) + ) + + return findings, checked_count, selected diff --git a/packages/tff-core/tests/test_adapter.py b/packages/tff-core/tests/test_adapter.py new file mode 100644 index 0000000..73f6d18 --- /dev/null +++ b/packages/tff-core/tests/test_adapter.py @@ -0,0 +1,384 @@ +"""Unit tests for PipelineAdapter interface, registries, and concrete adapters.""" + +from pathlib import Path +from unittest.mock import MagicMock, patch +import pytest + +from tff.core.adapter import ( + PipelineAdapter, + detect_provider, + get_adapter, +) +from tff.core.cli import _MockRunnerAdapter, _get_adapter, main +from tff.core.config import FitnessFunctionsConfig +from tff.dataform.adapter import DataformAdapter +from tff.dbt.adapter import DBTAdapter +from tff.sqlmesh.adapter import SQLMeshAdapter + + +class ConcreteDummyAdapter(PipelineAdapter): + @property + def provider_name(self) -> str: + return "dummy" + + def is_applicable(self, project_root: Path) -> bool: + return True + + def load_models( + self, + project_root: Path, + dialect: str | None = None, + manifest_path: str | Path | None = None, + ): + return {} + + def run_checks( + self, + project_root: Path, + config, + checks=None, + dialect=None, + manifest_path=None, + models=None, + ): + return [], 0, [] + + +def test_pipeline_adapter_base_defaults(): + adapter = ConcreteDummyAdapter() + assert adapter.provider_name == "dummy" + assert adapter.is_applicable(Path.cwd()) is True + assert adapter.load_models(Path.cwd()) == {} + assert adapter.run_checks(Path.cwd(), MagicMock()) == ([], 0, []) + assert ( + adapter.apply_metadata_fix(Path.cwd(), Path("foo.sql"), "foo", True, True) + is None + ) + assert adapter.get_diagnostic_files(Path.cwd()) == [] + + +def test_get_adapter_instances(): + assert isinstance(get_adapter("dbt"), DBTAdapter) + assert isinstance(get_adapter("sqlmesh"), SQLMeshAdapter) + assert isinstance(get_adapter("dataform"), DataformAdapter) + + +def test_custom_registered_adapter(): + from tff.core.adapter import ADAPTER_CLASSES + + ADAPTER_CLASSES["concrete_dummy"] = ("test_adapter", "ConcreteDummyAdapter") + try: + adapter = get_adapter("concrete_dummy") + assert isinstance(adapter, ConcreteDummyAdapter) + finally: + ADAPTER_CLASSES.pop("concrete_dummy", None) + + +def test_get_adapter_unknown(): + with pytest.raises(ValueError, match="Unknown provider: unknown"): + get_adapter("unknown") + + +def test_get_adapter_import_errors(): + with patch("importlib.import_module", side_effect=ImportError("mocked dbt")): + with pytest.raises(ImportError, match="tff-core\\[dbt\\]"): + get_adapter("dbt") + + with patch("importlib.import_module", side_effect=ImportError("mocked sqlmesh")): + with pytest.raises(ImportError, match="tff-core\\[sqlmesh\\]"): + get_adapter("sqlmesh") + + with patch("importlib.import_module", side_effect=ImportError("mocked dataform")): + with pytest.raises(ImportError, match="tff-core\\[dataform\\]"): + get_adapter("dataform") + + +def test_detect_provider(tmp_path: Path): + # Empty + with pytest.raises(ValueError, match="Could not detect project type"): + detect_provider(tmp_path) + + # dbt + (tmp_path / "dbt_project.yml").touch() + assert detect_provider(tmp_path) == "dbt" + + # dbt + sqlmesh (ambiguous) + (tmp_path / "config.py").touch() + with pytest.raises(ValueError, match="Both dbt and SQLMesh"): + detect_provider(tmp_path) + + # Clean dbt + (tmp_path / "dbt_project.yml").unlink() + assert detect_provider(tmp_path) == "sqlmesh" + + # Clean sqlmesh + (tmp_path / "config.py").unlink() + + # sqlmesh alternatives + for sig in (".sqlmesh", "config.yaml", "config.yml"): + p = tmp_path / sig + p.touch() + assert detect_provider(tmp_path) == "sqlmesh" + p.unlink() + + # dataform + (tmp_path / "workflow_settings.yaml").touch() + assert detect_provider(tmp_path) == "dataform" + (tmp_path / "workflow_settings.yaml").unlink() + + (tmp_path / "dataform.json").touch() + assert detect_provider(tmp_path) == "dataform" + + # multiple detected (e.g. dataform and sqlmesh) + (tmp_path / "config.yaml").touch() + with pytest.raises(ValueError, match="Multiple pipeline configuration files"): + detect_provider(tmp_path) + + +def test_dbt_adapter(tmp_path: Path): + adapter = DBTAdapter() + assert adapter.provider_name == "dbt" + + assert not adapter.is_applicable(tmp_path) + (tmp_path / "dbt_project.yml").touch() + assert adapter.is_applicable(tmp_path) + + with patch( + "tff.dbt.manifest.load_dbt_models", return_value={"m": MagicMock()} + ) as mock_load: + models = adapter.load_models(tmp_path, dialect="duckdb") + assert "m" in models + mock_load.assert_called_once_with(tmp_path, dialect="duckdb") + + cfg = FitnessFunctionsConfig() + with patch( + "tff.dbt.runner.run_all_checks", return_value=([], 1, ["rules"]) + ) as mock_run: + res = adapter.run_checks( + tmp_path, cfg, checks=["rules"], dialect="duckdb", models=models + ) + assert res == ([], 1, ["rules"]) + mock_run.assert_called_once_with( + project_root=tmp_path, + config=cfg, + checks=["rules"], + dialect="duckdb", + models=models, + ) + + with patch("tff.core.autofix.fix_dbt_metadata", return_value="fixed") as mock_fix: + res = adapter.apply_metadata_fix( + tmp_path, tmp_path / "models/m.sql", "m", True, False + ) + assert res == "fixed" + mock_fix.assert_called_once() + + diag = adapter.get_diagnostic_files(tmp_path) + assert len(diag) == 2 + assert diag[0][0] == "dbt_project.yml" + assert "found" in diag[0][1] + assert diag[1][0] == "manifest.json" + assert "missing" in diag[1][1] + + +def test_sqlmesh_adapter(tmp_path: Path): + adapter = SQLMeshAdapter() + assert adapter.provider_name == "sqlmesh" + + assert not adapter.is_applicable(tmp_path) + (tmp_path / "config.py").touch() + assert adapter.is_applicable(tmp_path) + + with ( + patch("sqlmesh.core.context.Context") as mock_ctx, + patch( + "tff.sqlmesh.runner.map_sqlmesh_context_models", + return_value={"m": MagicMock()}, + ), + ): + models = adapter.load_models(tmp_path) + assert "m" in models + mock_ctx.assert_called_once() + + cfg = FitnessFunctionsConfig() + with patch( + "tff.sqlmesh.runner.run_all_checks", return_value=([], 1, ["sqlmesh"]) + ) as mock_run: + res = adapter.run_checks(tmp_path, cfg, checks=["sqlmesh"], models=models) + assert res == ([], 1, ["sqlmesh"]) + mock_run.assert_called_once_with( + project_root=tmp_path, + config=cfg, + checks=["sqlmesh"], + models=models, + ) + + with patch( + "tff.core.autofix.fix_sqlmesh_metadata", return_value="fixed" + ) as mock_fix: + res = adapter.apply_metadata_fix( + tmp_path, tmp_path / "models/m.sql", "m", True, False + ) + assert res == "fixed" + mock_fix.assert_called_once() + + diag = adapter.get_diagnostic_files(tmp_path) + assert len(diag) == 2 + assert diag[0][0] == "config.py" + assert "found" in diag[0][1] + assert diag[1][0] == "settings.yaml" + assert "missing" in diag[1][1] + + +def test_dataform_adapter(tmp_path: Path): + adapter = DataformAdapter() + assert adapter.provider_name == "dataform" + + assert not adapter.is_applicable(tmp_path) + (tmp_path / "workflow_settings.yaml").touch() + assert adapter.is_applicable(tmp_path) + + with patch( + "tff.dataform.manifest.load_dataform_models", return_value={"m": MagicMock()} + ) as mock_load: + models = adapter.load_models( + tmp_path, dialect="bigquery", manifest_path=Path("manifest.json") + ) + assert "m" in models + mock_load.assert_called_once_with( + project_root=tmp_path, + manifest_path=Path("manifest.json"), + dialect="bigquery", + ) + + cfg = FitnessFunctionsConfig() + with patch( + "tff.dataform.runner.run_all_checks", return_value=([], 1, ["rules"]) + ) as mock_run: + res = adapter.run_checks( + tmp_path, + cfg, + checks=["rules"], + dialect="bigquery", + manifest_path="manifest.json", + models=models, + ) + assert res == ([], 1, ["rules"]) + mock_run.assert_called_once_with( + project_root=tmp_path, + config=cfg, + checks=["rules"], + dialect="bigquery", + manifest_path="manifest.json", + models=models, + ) + + assert ( + adapter.apply_metadata_fix( + tmp_path, tmp_path / "definitions/m.sqlx", "m", True, False + ) + is None + ) + + diag = adapter.get_diagnostic_files(tmp_path) + assert len(diag) == 2 + assert diag[0][0] == "workflow_settings.yaml" + assert "found" in diag[0][1] + assert diag[1][0] == "compilation manifest" + + # dataform.json diagnostic branch + (tmp_path / "workflow_settings.yaml").unlink() + (tmp_path / "dataform.json").touch() + diag2 = adapter.get_diagnostic_files(tmp_path) + assert diag2[0][0] == "dataform.json" + assert "found" in diag2[0][1] + + # missing diagnostic branch + (tmp_path / "dataform.json").unlink() + diag3 = adapter.get_diagnostic_files(tmp_path) + assert diag3[0][0] == "workflow_settings.yaml" + assert "missing" in diag3[0][1] + + +def test_mock_runner_adapter_all_methods(tmp_path: Path): + mock_runner = MagicMock() + mock_runner.run_all_checks.return_value = ([], 3, ["rules"]) + + # dbt + dbt_adapter = _MockRunnerAdapter("dbt", mock_runner) + assert dbt_adapter.provider_name == "dbt" + assert dbt_adapter.is_applicable(tmp_path) is True + with patch("tff.dbt.manifest.load_dbt_models", return_value={"m1": MagicMock()}): + assert "m1" in dbt_adapter.load_models(tmp_path, dialect="duckdb") + cfg = FitnessFunctionsConfig() + assert dbt_adapter.run_checks( + tmp_path, cfg, dialect="duckdb", models={"m1": MagicMock()} + ) == ([], 3, ["rules"]) + with patch("tff.core.autofix.fix_dbt_metadata", return_value="fixed"): + assert ( + dbt_adapter.apply_metadata_fix( + tmp_path, tmp_path / "models/m.sql", "m", True, True + ) + == "fixed" + ) + assert len(dbt_adapter.get_diagnostic_files(tmp_path)) == 2 + + # sqlmesh + sqlmesh_adapter = _MockRunnerAdapter("sqlmesh", mock_runner) + assert sqlmesh_adapter.provider_name == "sqlmesh" + with ( + patch("sqlmesh.core.context.Context"), + patch( + "tff.sqlmesh.runner.map_sqlmesh_context_models", + return_value={"m2": MagicMock()}, + ), + ): + assert "m2" in sqlmesh_adapter.load_models(tmp_path) + assert sqlmesh_adapter.run_checks(tmp_path, cfg, models={"m2": MagicMock()}) == ( + [], + 3, + ["rules"], + ) + with patch("tff.core.autofix.fix_sqlmesh_metadata", return_value="fixed_sm"): + assert ( + sqlmesh_adapter.apply_metadata_fix( + tmp_path, tmp_path / "models/m.sql", "m", True, True + ) + == "fixed_sm" + ) + assert len(sqlmesh_adapter.get_diagnostic_files(tmp_path)) == 2 + + # dataform + df_adapter = _MockRunnerAdapter("dataform", mock_runner) + assert df_adapter.provider_name == "dataform" + with patch( + "tff.dataform.manifest.load_dataform_models", return_value={"m3": MagicMock()} + ): + assert "m3" in df_adapter.load_models(tmp_path) + assert df_adapter.run_checks( + tmp_path, + cfg, + dialect="bigquery", + manifest_path="m.json", + models={"m3": MagicMock()}, + ) == ([], 3, ["rules"]) + + # other/unknown provider + other_adapter = _MockRunnerAdapter("other", mock_runner) + assert other_adapter.load_models(tmp_path) == {} + assert ( + other_adapter.apply_metadata_fix(tmp_path, tmp_path / "x", "m", True, True) + is None + ) + assert other_adapter.get_diagnostic_files(tmp_path) == [] + + +def test_cli_info_adapter_error(tmp_path: Path): + with patch( + "tff.core.cli._get_adapter", side_effect=ValueError("Test adapter error") + ): + assert main(["info", "--project", str(tmp_path), "--provider", "dbt"]) == 1 + + +def test_cli_get_adapter(): + assert isinstance(_get_adapter("dbt"), DBTAdapter)