Merge pull request #62 from FreshCode-Org/feat/quality-ops-interop

Quality-Ops interoperability: dbt / Great Expectations / exception tables / lineage exporters

Author: JohnnyWilson16

src/freshdata/__init__.py

@@ -39,6 +39,7 @@
 from .config import CleanConfig
 from .execution import EngineConfig
 from .explain import ExplainReport, explain_clean
+from .findings import QualityFinding
 from .plan import CleanPlan, ColumnPlan, compare_clean, compare_plans
 from .profile import ColumnProfile, Profile
 from .report import Action, CleanReport
@@ -59,6 +60,7 @@
     "EngineConfig",
     "ExplainReport",
     "Profile",
+    "QualityFinding",
     "StreamingCleanConfig",
     "StreamingCleaner",
     "StreamingState",
@@ -135,6 +137,18 @@
     "BlockingRule",
 })
 
+#: Quality-ops exporters served lazily from :mod:`freshdata.integrations`. Like the
+#: enterprise exports they are kept out of ``__all__`` so ``import freshdata`` stays
+#: light — importing the integrations layer pulls in the enterprise layer and its
+#: optional deps, which should only happen when an exporter is actually used.
+_INTEGRATION_EXPORTS = {
+    "export_quality_ops": "freshdata.integrations.quality_ops",
+    "QualityOpsResult": "freshdata.integrations.quality_ops",
+    "export_dbt_tests": "freshdata.integrations.dbt",
+    "export_gx_suite": "freshdata.integrations.great_expectations",
+    "build_exception_table": "freshdata.integrations.exceptions",
+}
+
 
 def __getattr__(name: str) -> object:
     """Lazily resolve the ``enterprise`` submodule and its key exports (PEP 562)."""
@@ -146,8 +160,12 @@ def __getattr__(name: str) -> object:
         import importlib
 
         return getattr(importlib.import_module("freshdata.enterprise"), name)
+    if name in _INTEGRATION_EXPORTS:
+        import importlib
+
+        return getattr(importlib.import_module(_INTEGRATION_EXPORTS[name]), name)
     raise AttributeError(f"module 'freshdata' has no attribute {name!r}")
 
 
 def __dir__() -> list:
-    return sorted([*__all__, "enterprise", *_ENTERPRISE_EXPORTS])
+    return sorted([*__all__, "enterprise", *_ENTERPRISE_EXPORTS, *_INTEGRATION_EXPORTS])

src/freshdata/enterprise/cli.py

@@ -15,6 +15,7 @@
 
 import argparse
 import json
+from pathlib import Path
 from typing import Any
 
 import pandas as pd
@@ -153,6 +154,42 @@ def cmd_trust(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_quality_ops(args: argparse.Namespace) -> int:
+    from ..findings import findings_from_dict
+    from ..integrations.quality_ops import export_quality_ops
+
+    with open(args.input, encoding="utf-8") as fh:
+        report_dict = json.load(fh)
+    findings = findings_from_dict(report_dict)
+
+    df = _read_frame(args.data, None) if args.data else None
+    model_name = args.model_name or Path(args.input).stem
+    suite_name = args.suite_name or f"{model_name}_suite"
+
+    result = export_quality_ops(
+        findings,
+        model_name=model_name,
+        suite_name=suite_name,
+        dbt_path=args.dbt,
+        gx_path=args.gx,
+        exception_table_path=args.exceptions,
+        df=df,
+        include_pii=args.include_pii,
+        exceptions_format=args.exceptions_format,
+    )
+    if args.lineage:
+        with open(args.lineage, "w", encoding="utf-8") as fh:
+            json.dump(result.lineage_event, fh, indent=2, default=str)
+    if not args.quiet:
+        print(f"freshdata quality-ops: {len(result.findings)} finding(s)")
+        for label, dest in (("dbt", result.dbt_path), ("gx", result.gx_path),
+                            ("exceptions", result.exception_table_path),
+                            ("lineage", args.lineage)):
+            if dest:
+                print(f"  {label}: {dest}")
+    return 0
+
+
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(prog="freshdata", description="freshdata enterprise CLI")
     subparsers = parser.add_subparsers(dest="command", required=True)
@@ -190,6 +227,28 @@ def build_parser() -> argparse.ArgumentParser:
                        help="exit non-zero if the trust score is below this")
     trust.set_defaults(func=cmd_trust)
 
+    qops = subparsers.add_parser(
+        "quality-ops",
+        help="export findings from a report.json to dbt/GX/exception/lineage artifacts",
+    )
+    qops.add_argument("input", help="path to a freshdata report JSON (CleanReport.to_dict)")
+    qops.add_argument("--dbt", metavar="schema.yml", help="write dbt generic tests YAML here")
+    qops.add_argument("--gx", metavar="suite.json", help="write a Great Expectations suite here")
+    qops.add_argument("--exceptions", metavar="PATH",
+                      help="write an exception table here (.csv/.parquet/.duckdb)")
+    qops.add_argument("--exceptions-format", choices=("csv", "parquet", "duckdb"),
+                      help="exception-table format (else inferred from the extension)")
+    qops.add_argument("--lineage", metavar="lineage.json",
+                      help="write the OpenLineage event (with artifact facets) here")
+    qops.add_argument("--data", metavar="PATH",
+                      help="optional source file to enrich exception observed_values")
+    qops.add_argument("--model-name", help="dbt model name (default: report filename stem)")
+    qops.add_argument("--suite-name", help="GX suite name (default: <model>_suite)")
+    qops.add_argument("--include-pii", action="store_true",
+                      help="reveal observed values in the exception table (default: redacted)")
+    qops.add_argument("--quiet", action="store_true")
+    qops.set_defaults(func=cmd_quality_ops)
+
     from ..streaming._cli import add_stream_subparsers
 
     add_stream_subparsers(subparsers)

src/freshdata/enterprise/contracts.py

@@ -41,6 +41,7 @@
 import pandas as pd
 
 from ..adapters.polars import to_pandas
+from ..findings import QualityFinding
 from .config import (
     AnonymizationConfig,  # noqa: F401  (re-exported for discoverability)
     DriftConfig,
@@ -464,6 +465,34 @@ def to_dict(self) -> dict[str, Any]:
             "contract_results": self.contract_results,
         }
 
+    def to_findings(self, *, lineage_run_id: str | None = None) -> list:
+        """Project warned/failed drift findings into :class:`~freshdata.QualityFinding`."""
+        out: list = []
+        for f in self.findings:
+            if f.status == "passed":
+                continue
+            expected = None
+            if f.metric is not None:
+                expected = str(f.metric)
+                if f.baseline_value is not None:
+                    expected += f" ~ baseline {f.baseline_value}"
+                if f.threshold is not None:
+                    expected += f" (threshold {f.threshold})"
+            out.append(QualityFinding.create(
+                severity=f.level,
+                step="drift",
+                column=f.column,
+                rule_name=f.check_id,
+                message=f.message,
+                observed_value=f.current_value,
+                expected_condition=expected,
+                action_taken=f.status,
+                lineage_run_id=lineage_run_id,
+                extra={"metric": f.metric, "baseline_value": f.baseline_value,
+                       "threshold": f.threshold, **(f.details or {})},
+            ))
+        return out
+
     def to_json(self, *, indent: int | None = 2) -> str:
         return json.dumps(self.to_dict(), indent=indent, default=str, sort_keys=True)
 

src/freshdata/enterprise/entity_resolution.py

@@ -228,6 +228,35 @@ def to_dict(self) -> dict[str, Any]:
             "runtime_metadata": self.runtime_metadata,
         }
 
+    def to_findings(self, *, lineage_run_id: str | None = None) -> list:
+        """Project match / possible-match pairs into :class:`~freshdata.QualityFinding`.
+
+        Each surviving pair is a candidate duplicate: ``match`` maps to ``error``,
+        ``possible_match`` to ``warning``; non-matches are dropped.
+        """
+        from ..findings import QualityFinding
+
+        out: list = []
+        for p in self.pairs:
+            if p.decision == "non_match":
+                continue
+            out.append(QualityFinding.create(
+                severity=p.decision,
+                step="entity_resolution",
+                column=None,
+                rule_name="duplicate_match",
+                message=(f"records {p.left_id} & {p.right_id} {p.decision} "
+                         f"(p={p.match_probability:.3f})"),
+                row_selector=f"{p.left_id} <-> {p.right_id}",
+                observed_value=p.comparison_vector,
+                expected_condition="distinct entities",
+                action_taken=p.decision,
+                lineage_run_id=lineage_run_id,
+                extra={"match_probability": round(p.match_probability, 4),
+                       "match_weight": round(p.match_weight, 4)},
+            ))
+        return out
+
     def summary(self) -> str:
         return (
             f"entity resolution ({self.backend}): {self.n_records} record(s), "

src/freshdata/enterprise/lineage.py

@@ -112,6 +112,13 @@ def __init__(
         self.output_name = output_name
         self.events: list[LineageEvent] = []
         self.started_at = _now_iso()
+        #: Extra OpenLineage run facets attached by callers (e.g. quality-ops artifact
+        #: paths). Merged into every emitted event alongside ``freshdata_transformations``.
+        self.extra_run_facets: dict[str, Any] = {}
+
+    def add_run_facet(self, name: str, facet: dict[str, Any]) -> None:
+        """Attach a custom OpenLineage run facet (e.g. ``"dbt_tests_path"``)."""
+        self.extra_run_facets[name] = facet
 
     def record(
         self,
@@ -204,7 +211,8 @@ def _event(self, event_type: str, *, with_facets: bool) -> dict[str, Any]:
                     }
                     for e in self.events
                 ],
-            }
+            },
+            **self.extra_run_facets,
         }
         output_facets: dict[str, Any] = {}
         input_facets: dict[str, Any] = {}

src/freshdata/enterprise/privacy.py

@@ -638,6 +638,48 @@ def to_dict(self) -> dict[str, Any]:
             "metadata": self.metadata,
         }
 
+    def to_findings(self, *, lineage_run_id: str | None = None) -> list:
+        """Project masking events (and any k-anonymity breach) into findings.
+
+        ``observed_value`` is the already-redacted ``original_preview`` and the
+        findings are flagged ``sensitive``, so PII never leaks through an export
+        unless the caller explicitly opts in.
+        """
+        from ..findings import QualityFinding
+
+        out: list = []
+        for e in self.events:
+            out.append(QualityFinding.create(
+                severity=e.risk_level,
+                step="privacy",
+                column=e.column,
+                rule_name=e.entity_type,
+                message=f"{e.entity_type} detected via {e.source}",
+                row_index=e.row,
+                observed_value=e.original_preview,
+                expected_condition="no PII",
+                action_taken=e.strategy,
+                lineage_run_id=lineage_run_id,
+                sensitive=True,
+                extra={"reversible": e.reversible, "format_preserving": e.format_preserving,
+                       "hipaa_tag": e.hipaa_tag, "gdpr_tag": e.gdpr_tag, "score": e.score},
+            ))
+        ka = self.k_anonymity or {}
+        if ka and not ka.get("ok", True):
+            out.append(QualityFinding.create(
+                severity="error",
+                step="privacy",
+                column=None,
+                rule_name="k_anonymity",
+                message=(f"k-anonymity violated: {ka.get('rows_violating_k')} row(s) "
+                         f"below k={ka.get('k')}"),
+                row_selector=f"quasi_identifiers: {list(ka.get('quasi_identifiers', []))}",
+                expected_condition=f"every group >= k={ka.get('k')}",
+                lineage_run_id=lineage_run_id,
+                extra={"violation_ratio": ka.get("violation_ratio")},
+            ))
+        return out
+
     def to_json(self, *, indent: int | None = 2) -> str:
         return json.dumps(self.to_dict(), indent=indent, default=str)