From 27a1f9481df94e56fa995a4f69042a765d0b4ddc Mon Sep 17 00:00:00 2001 From: Richard Iannone Date: Tue, 11 Aug 2026 00:07:59 -0400 Subject: [PATCH 1/5] Add DataScan serialization and diff utilities --- pointblank/datascan.py | 455 ++++++++++++++++++++++++++++++++++++++++- 1 file changed, 450 insertions(+), 5 deletions(-) diff --git a/pointblank/datascan.py b/pointblank/datascan.py index 0754202e2..73dfb3d25 100644 --- a/pointblank/datascan.py +++ b/pointblank/datascan.py @@ -2,6 +2,7 @@ import contextlib import json +from dataclasses import dataclass from importlib.metadata import version from typing import TYPE_CHECKING, Any, cast @@ -22,7 +23,7 @@ from pointblank.scan_profile_stats import StatGroup -__all__ = ["DataScan", "col_summary_tbl"] +__all__ = ["DataScan", "DataScanDiff", "col_summary_tbl"] class DataScan: @@ -487,15 +488,459 @@ def _build_label_map(cols: Sequence[str]) -> dict[str, Any]: label_map[target_col] = matching_stat.label return label_map + def to_dict(self) -> dict[str, Any]: + """ + Export the profile as a structured dictionary. + + The returned dictionary contains metadata (table name, row count, column list) plus + per-column profile entries with their data type, statistics, and sample data. This format is + designed for round-trip persistence: save it with `to_json()` / `save_to_json()` and restore + with `from_dict()` / `from_json()` / `load_from_json()`. + + Returns + ------- + dict[str, Any] + A dictionary with keys `"metadata"` and `"columns"`. + """ + columns_out: list[dict[str, Any]] = [] + for prof in self.profile.column_profiles: + stat_dict: dict[str, Any] = {} + for stat in prof.statistics: + stat_dict[stat.name] = stat.val + + columns_out.append( + { + "colname": prof.colname, + "coltype": prof.coltype, + "sample_data": list(prof.sample_data), + "statistics": stat_dict, + } + ) + + return { + "metadata": { + "table_name": self.profile.table_name, + "row_count": self.profile.row_count, + "columns": self.profile.columns, + }, + "columns": columns_out, + } + def to_json(self) -> str: - prof_dict = self.profile.as_dataframe(strict=False).to_dict(as_series=False) + """ + Export the profile as a JSON string. - return json.dumps(prof_dict, indent=4, default=str) + The JSON is structured for round-trip persistence. Use `from_json()` or `load_from_json()` + to restore a `DataScan` from the output. + + Returns + ------- + str + A JSON string representing the profile. + """ + return json.dumps(self.to_dict(), indent=4, default=str) def save_to_json(self, output_file: str) -> None: - json_string: str = self.to_json() + """ + Save the profile to a JSON file. + + Parameters + ---------- + output_file + The path to the output JSON file. + """ with open(output_file, "w") as f: - json.dump(json_string, f, indent=4) + f.write(self.to_json()) + + @classmethod + def from_dict(cls, d: dict[str, Any]) -> DataScan: + """ + Restore a `DataScan` from a dictionary produced by `to_dict()`. + + This reconstructs the profile without needing the original data. + + Parameters + ---------- + d + A dictionary with `"metadata"` and `"columns"` keys, as produced by `to_dict()`. + + Returns + ------- + DataScan + A restored `DataScan` instance. + """ + meta = d["metadata"] + col_entries = d["columns"] + + obj = cls.__new__(cls) + obj.nw_data = None # type: ignore[assignment] + obj.tbl_name = meta.get("table_name") + + profile = _DataProfile.__new__(_DataProfile) + profile.table_name = meta.get("table_name") + profile.row_count = meta["row_count"] + profile.columns = meta["columns"] + profile.implementation = nw.Implementation.POLARS + profile.column_profiles = [] + + stat_class_map: dict[str, type] = { + stat_cls.name: stat_cls for stat_cls in COLUMN_ORDER_REGISTRY + } + + for col_entry in col_entries: + col_prof = ColumnProfile( + colname=col_entry["colname"], + coltype=col_entry["coltype"], + ) + col_prof.sample_data = col_entry.get("sample_data", []) + + for stat_name, stat_val in col_entry.get("statistics", {}).items(): + stat_cls = stat_class_map.get(stat_name) + if stat_cls is not None: + col_prof.statistics.append(stat_cls(stat_val)) + + _assign_type_from_coltype(col_prof) + profile.column_profiles.append(col_prof) + + obj.profile = profile + return obj + + @classmethod + def from_json(cls, json_string: str) -> DataScan: + """ + Restore a `DataScan` from a JSON string produced by `to_json()`. + + Parameters + ---------- + json_string + A JSON string as produced by `to_json()`. + + Returns + ------- + DataScan + A restored `DataScan` instance. + """ + return cls.from_dict(json.loads(json_string)) + + @classmethod + def load_from_json(cls, input_file: str) -> DataScan: + """ + Load a `DataScan` from a JSON file produced by `save_to_json()`. + + Parameters + ---------- + input_file + The path to the JSON file. + + Returns + ------- + DataScan + A restored `DataScan` instance. + """ + with open(input_file) as f: + return cls.from_json(f.read()) + + def compare(self, baseline: DataScan) -> DataScanDiff: + """ + Compare this scan against a baseline and return the differences. + + The comparison covers schema changes (columns added, removed, or with changed types) and + statistical drift for columns present in both scans. The returned `DataScanDiff` object + provides programmatic access to the results and a tabular report via `get_tabular_report()`. + + Parameters + ---------- + baseline + The baseline `DataScan` to compare against (typically the older scan). + + Returns + ------- + DataScanDiff + An object describing the differences between the two scans. + """ + return DataScanDiff(current=self, baseline=baseline) + + +def _assign_type_from_coltype(prof: ColumnProfile) -> None: + coltype_lower = prof.coltype.lower() + for type_enum in _TypeMap: + if any(ind in coltype_lower for ind in type_enum.value): + prof.__class__ = _TypeMap.fetch_prof_map()[type_enum] + return + + +@dataclass +class _ColumnDiff: + colname: str + coltype_baseline: str | None + coltype_current: str | None + status: str + stat_diffs: dict[str, tuple[Any, Any]] + + +class DataScanDiff: + """ + The result of comparing two `DataScan` profiles. + + Created by calling `DataScan.compare()`. Provides programmatic access to schema changes and + per-column statistical drift, plus a tabular report via `get_tabular_report()`. + + Attributes + ---------- + columns_added + Column names present in the current scan but not the baseline. + columns_removed + Column names present in the baseline but not the current scan. + columns_type_changed + Column names whose data type changed between baseline and current. + column_diffs + Per-column diff details for all columns that appear in either scan. + """ + + def __init__(self, current: DataScan, baseline: DataScan) -> None: + self._current = current + self._baseline = baseline + + cur_profiles = {p.colname: p for p in current.profile.column_profiles} + base_profiles = {p.colname: p for p in baseline.profile.column_profiles} + + cur_names = set(cur_profiles.keys()) + base_names = set(base_profiles.keys()) + + self.columns_added: list[str] = sorted(cur_names - base_names) + self.columns_removed: list[str] = sorted(base_names - cur_names) + + self.columns_type_changed: list[str] = [] + self.column_diffs: list[_ColumnDiff] = [] + + all_col_names = list(dict.fromkeys(list(base_profiles.keys()) + list(cur_profiles.keys()))) + + for col_name in all_col_names: + base_prof = base_profiles.get(col_name) + cur_prof = cur_profiles.get(col_name) + + if base_prof is None: + self.column_diffs.append( + _ColumnDiff( + colname=col_name, + coltype_baseline=None, + coltype_current=cur_prof.coltype if cur_prof else None, + status="added", + stat_diffs={}, + ) + ) + continue + + if cur_prof is None: + self.column_diffs.append( + _ColumnDiff( + colname=col_name, + coltype_baseline=base_prof.coltype, + coltype_current=None, + status="removed", + stat_diffs={}, + ) + ) + continue + + type_changed = base_prof.coltype != cur_prof.coltype + if type_changed: + self.columns_type_changed.append(col_name) + + base_stats = {s.name: s.val for s in base_prof.statistics} + cur_stats = {s.name: s.val for s in cur_prof.statistics} + all_stat_names = list(dict.fromkeys(list(base_stats.keys()) + list(cur_stats.keys()))) + + stat_diffs: dict[str, tuple[Any, Any]] = {} + for stat_name in all_stat_names: + base_val = base_stats.get(stat_name) + cur_val = cur_stats.get(stat_name) + if base_val != cur_val: + stat_diffs[stat_name] = (base_val, cur_val) + + status = "type_changed" if type_changed else ("changed" if stat_diffs else "unchanged") + self.column_diffs.append( + _ColumnDiff( + colname=col_name, + coltype_baseline=base_prof.coltype, + coltype_current=cur_prof.coltype, + status=status, + stat_diffs=stat_diffs, + ) + ) + + @property + def has_changes(self) -> bool: + """Return ``True`` if any schema or statistical changes were detected.""" + return bool( + self.columns_added + or self.columns_removed + or self.columns_type_changed + or any(d.stat_diffs for d in self.column_diffs) + ) + + @property + def row_count_diff(self) -> tuple[int, int]: + """Return `(baseline_row_count, current_row_count)`.""" + return (self._baseline.profile.row_count, self._current.profile.row_count) + + def to_dict(self) -> dict[str, Any]: + """ + Export the comparison results as a dictionary. + + Returns + ------- + dict[str, Any] + A dictionary with schema changes, row count diff, and per-column stat diffs. + """ + return { + "row_count": { + "baseline": self._baseline.profile.row_count, + "current": self._current.profile.row_count, + }, + "columns_added": self.columns_added, + "columns_removed": self.columns_removed, + "columns_type_changed": [ + { + "column": d.colname, + "baseline_type": d.coltype_baseline, + "current_type": d.coltype_current, + } + for d in self.column_diffs + if d.status == "type_changed" + ], + "stat_diffs": { + d.colname: { + stat_name: {"baseline": bv, "current": cv} + for stat_name, (bv, cv) in d.stat_diffs.items() + } + for d in self.column_diffs + if d.stat_diffs + }, + } + + def get_tabular_report(self) -> GT: + """ + Generate a GT table summarizing the differences between the two scans. + + Returns + ------- + GT + A styled Great Tables report showing schema and statistical drift. + """ + import polars as pl + + rows: list[dict[str, Any]] = [] + + for diff in self.column_diffs: + if diff.status == "added": + rows.append( + { + "column": diff.colname, + "status": "Added", + "type_baseline": "", + "type_current": diff.coltype_current or "", + "stat_changes": "", + } + ) + elif diff.status == "removed": + rows.append( + { + "column": diff.colname, + "status": "Removed", + "type_baseline": diff.coltype_baseline or "", + "type_current": "", + "stat_changes": "", + } + ) + else: + status = "Type Changed" if diff.status == "type_changed" else "OK" + if diff.stat_diffs: + status = ( + "Type + Stats Changed" if diff.status == "type_changed" else "Stats Changed" + ) + + change_parts: list[str] = [] + for stat_name, (bv, cv) in diff.stat_diffs.items(): + if stat_name == "freqs": + continue + bv_str = _format_stat_value(bv) + cv_str = _format_stat_value(cv) + change_parts.append(f"{stat_name}: {bv_str} -> {cv_str}") + + rows.append( + { + "column": diff.colname, + "status": status, + "type_baseline": diff.coltype_baseline or "", + "type_current": diff.coltype_current or "", + "stat_changes": "; ".join(change_parts), + } + ) + + if not rows: + rows.append( + { + "column": "(no columns)", + "status": "OK", + "type_baseline": "", + "type_current": "", + "stat_changes": "", + } + ) + + df = pl.DataFrame(rows) + + base_rc, cur_rc = self.row_count_diff + rc_note = f"Row count: {base_rc:,} (baseline) vs {cur_rc:,} (current)" + + base_name = self._baseline.tbl_name or "baseline" + cur_name = self._current.tbl_name or "current" + + gt_tbl = ( + GT(df) + .tab_header( + title=html(f"Profile Comparison: {base_name} vs {cur_name}"), + subtitle=html(rc_note), + ) + .cols_label( + column="Column", + status="Status", + type_baseline="Type (Baseline)", + type_current="Type (Current)", + stat_changes="Changed Statistics", + ) + .opt_table_font(font=google_font("IBM Plex Sans")) + .opt_align_table_header(align="left") + .tab_style( + style=style.text(font=google_font("IBM Plex Mono")), + locations=loc.body(), + ) + .tab_style( + style=style.text(size="11px"), + locations=loc.body(columns="stat_changes"), + ) + ) + + return gt_tbl + + def __repr__(self) -> str: + n_changed = sum(1 for d in self.column_diffs if d.status != "unchanged") + return ( + f"DataScanDiff(" + f"added={len(self.columns_added)}, " + f"removed={len(self.columns_removed)}, " + f"type_changed={len(self.columns_type_changed)}, " + f"stat_changed={n_changed - len(self.columns_added) - len(self.columns_removed)})" + ) + + +def _format_stat_value(val: Any) -> str: + if val is None: + return "-" + if isinstance(val, float): + return f"{val:.4g}" + return str(val) def col_summary_tbl(data: Any, tbl_name: str | None = None) -> GT: From d0c5962b0889e75c107ae1578a6cf9d9a91f4b96 Mon Sep 17 00:00:00 2001 From: Richard Iannone Date: Tue, 11 Aug 2026 00:08:03 -0400 Subject: [PATCH 2/5] Update __init__.py --- pointblank/__init__.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pointblank/__init__.py b/pointblank/__init__.py index 52f8fba31..b5a344020 100644 --- a/pointblank/__init__.py +++ b/pointblank/__init__.py @@ -32,7 +32,7 @@ starts_with, ) from pointblank.contract import Contract, Step -from pointblank.datascan import DataScan, col_summary_tbl +from pointblank.datascan import DataScan, DataScanDiff, col_summary_tbl from pointblank.draft import DraftValidation from pointblank.edit import EditValidation from pointblank.field import ( @@ -126,6 +126,7 @@ "Pipeline", "PipelineResult", "DataScan", + "DataScanDiff", "DraftValidation", "EditValidation", "MissingSpec", From 921366cc91504652424fa558d96ad67279ea8ad5 Mon Sep 17 00:00:00 2001 From: Richard Iannone Date: Tue, 11 Aug 2026 00:08:06 -0400 Subject: [PATCH 3/5] Update test_datascan.py --- tests/test_datascan.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/test_datascan.py b/tests/test_datascan.py index 49d39e9b4..7690c61a1 100644 --- a/tests/test_datascan.py +++ b/tests/test_datascan.py @@ -550,8 +550,9 @@ def test_datascan_save_to_json(tmp_path): with open(output_file) as f: content = json.load(f) - # The saved content should be a JSON string (due to json.dump with a string) - assert isinstance(content, str) + assert isinstance(content, dict) + assert "metadata" in content + assert "columns" in content def test_typemap_fetch_icon_with_unknown_type(): From 1cedc5de4ec5b0e81a6b38c44b6b1c6a8dbcc77a Mon Sep 17 00:00:00 2001 From: Richard Iannone Date: Tue, 11 Aug 2026 00:08:09 -0400 Subject: [PATCH 4/5] Create test_datascan_persistence.py --- tests/test_datascan_persistence.py | 337 +++++++++++++++++++++++++++++ 1 file changed, 337 insertions(+) create mode 100644 tests/test_datascan_persistence.py diff --git a/tests/test_datascan_persistence.py b/tests/test_datascan_persistence.py new file mode 100644 index 000000000..f3439013f --- /dev/null +++ b/tests/test_datascan_persistence.py @@ -0,0 +1,337 @@ +import json + +import pandas as pd +import polars as pl +import pytest + +import pointblank as pb +from pointblank.datascan import DataScan, DataScanDiff + + +# ── Fixtures ────────────────────────────────────────────────────────────────── + + +@pytest.fixture +def sample_df(): + return pl.DataFrame( + { + "id": [1, 2, 3, None], + "name": ["Alice", "Bob", "Charlie", "Diana"], + "score": [95.5, 82.3, None, 91.0], + "active": [True, False, True, True], + } + ) + + +@pytest.fixture +def scan(sample_df): + return DataScan(data=sample_df, tbl_name="test_table") + + +# ── to_dict ─────────────────────────────────────────────────────────────────── + + +def test_to_dict_structure(scan): + d = scan.to_dict() + assert "metadata" in d + assert "columns" in d + assert d["metadata"]["table_name"] == "test_table" + assert d["metadata"]["row_count"] == 4 + assert d["metadata"]["columns"] == ["id", "name", "score", "active"] + assert len(d["columns"]) == 4 + + +def test_to_dict_column_entries(scan): + d = scan.to_dict() + col_entry = d["columns"][0] + assert "colname" in col_entry + assert "coltype" in col_entry + assert "sample_data" in col_entry + assert "statistics" in col_entry + assert isinstance(col_entry["statistics"], dict) + + +def test_to_dict_no_svg_icons(scan): + d = scan.to_dict() + json_str = json.dumps(d, default=str) + assert " Date: Tue, 11 Aug 2026 00:08:20 -0400 Subject: [PATCH 5/5] Expand DataScan docs in summary guide --- .../05-data-inspection/02-col-summary-tbl.qmd | 109 +++++++++++++++++- 1 file changed, 108 insertions(+), 1 deletion(-) diff --git a/user_guide/05-data-inspection/02-col-summary-tbl.qmd b/user_guide/05-data-inspection/02-col-summary-tbl.qmd index db9067247..6c94a2b55 100644 --- a/user_guide/05-data-inspection/02-col-summary-tbl.qmd +++ b/user_guide/05-data-inspection/02-col-summary-tbl.qmd @@ -9,7 +9,7 @@ bread-crumbs: true ```{python} #| echo: false #| output: false -import pointblank as pb +import os ``` While previewing a table with `preview()` is undoubtedly a good thing to do, sometimes you need @@ -100,3 +100,110 @@ left of the column names) are: One column, `e`, is of the `Boolean` type. Because columns of this type could only have `True`, `False`, or missing values, we provide summary data for missingness (under `NA`) and proportions of `True` and `False` values (under `UQ`). + +## Using `DataScan` Directly + +The `col_summary_tbl()` function is a convenience wrapper around the +[`DataScan`](`pointblank.DataScan`) class. When you need more than a visual report (for example, to +save the profile for later comparison), you can work with `DataScan` directly: + +```{python} +small_table = pb.load_dataset(dataset="small_table", tbl_type="polars") + +scan = pb.DataScan(data=small_table, tbl_name="small_table") +``` + +The `DataScan` object computes the same column-level statistics shown in the summary table. You can +access the profile as a dictionary with `to_dict()`, export it as JSON with `to_json()`, or render +the same tabular report with `get_tabular_report()`. + +## Saving and Loading Profiles + +A `DataScan` profile can be saved to disk and loaded back later. This is useful for establishing +baselines that you compare against in the future. + +```{python} +# Save the profile to a JSON file +scan.save_to_json("small_table_profile.json") + +# Later, load it back without needing the original data +loaded = pb.DataScan.load_from_json("small_table_profile.json") +``` + +```{python} +#| echo: false +#| output: false +os.remove("small_table_profile.json") +``` + +The `from_json()` classmethod works with JSON strings directly, and `from_dict()` accepts the +dictionary format produced by `to_dict()`. All three approaches produce a fully restored `DataScan` +that retains the column names, types, statistics, and sample data from the original scan. + +## Comparing Profiles for Drift + +When your data changes over time, you can compare two `DataScan` profiles to detect drift. The +`compare()` method identifies schema changes (columns added, removed, or with changed types) and +statistical shifts in common columns. + +```{python} +import polars as pl + +# Simulate two versions of a dataset +orders_v1 = pl.DataFrame({ + "order_id": [1, 2, 3, 4, 5], + "amount": [10.0, 25.0, 15.0, 30.0, 20.0], + "status": ["paid", "paid", "refund", "paid", "paid"], +}) + +orders_v2 = pl.DataFrame({ + "order_id": [1, 2, 3, 4, 5, 6, 7, 8], + "amount": [10.0, 25.0, 15.0, 30.0, 20.0, 150.0, 200.0, 175.0], + "status": ["paid", "paid", "refund", "paid", "paid", "paid", "paid", "paid"], + "region": ["US", "EU", "US", "EU", "US", "US", "EU", "EU"], +}) + +baseline = pb.DataScan(data=orders_v1, tbl_name="orders_v1") +current = pb.DataScan(data=orders_v2, tbl_name="orders_v2") + +diff = current.compare(baseline) +``` + +The returned [`DataScanDiff`](`pointblank.DataScanDiff`) object provides programmatic access to the +changes: + +```{python} +print("Has changes:", diff.has_changes) +print("Columns added:", diff.columns_added) +print("Row count (baseline vs current):", diff.row_count_diff) +``` + +You can also get the full comparison as a dictionary with `to_dict()`, or view it as a styled +report: + +```{python} +diff.get_tabular_report() +``` + +The report shows each column's status (OK, Added, Removed, Stats Changed, or Type Changed) along +with any statistics that shifted between the baseline and current profiles. This makes it +straightforward to spot when your data's shape or distribution has changed in ways that might affect +downstream analyses or validation rules. + +A typical workflow is to save a baseline profile after your initial data quality checks pass, then +compare new data against that baseline on each pipeline run: + +```python +# On the first run: establish the baseline +baseline = pb.DataScan(data=production_table, tbl_name="orders") +baseline.save_to_json("orders_baseline.json") + +# On subsequent runs: compare against the baseline +baseline = pb.DataScan.load_from_json("orders_baseline.json") +current = pb.DataScan(data=production_table, tbl_name="orders") +diff = current.compare(baseline) + +if diff.has_changes: + print("Data drift detected!") + print(diff.to_dict()) +```