fix: accept file paths and connection strings for col_vals_in_table(ref_table=)

pointblank/validate.py

@@ -66,6 +66,7 @@
     NumberOfTestUnits,
     SpeciallyValidation,
     StepEvaluationError,
+    _first_line,
     apply_missing_exclusion,
     col_count_match,
     col_exists,
@@ -4684,13 +4685,18 @@ def _columns_param() -> None:
     elif at == "col_vals_in_table":
         values = vi.values or {}
         params["columns"] = values.get("columns", vi.column)
-        params["ref_table"] = _placeholder(
-            code="your_reference_table",
-            note=(
-                "Step 'col_vals_in_table' checks against a reference table that cannot be "
-                "serialized; a placeholder was emitted and must be edited manually."
-            ),
-        )
+        ref_table = values.get("ref_table")
+        if isinstance(ref_table, (str, Path)):
+            # A file path or connection string round-trips as-is
+            params["ref_table"] = str(ref_table)
+        else:
+            params["ref_table"] = _placeholder(
+                code="your_reference_table",
+                note=(
+                    "Step 'col_vals_in_table' checks against a reference table that cannot be "
+                    "serialized; a placeholder was emitted and must be edited manually."
+                ),
+            )
         params["ref_column"] = values.get("ref_column")
     elif at == "conjointly":
         params["expressions"] = _placeholder(
@@ -14858,8 +14864,12 @@ def col_vals_in_table(
             The column (or list of columns for composite keys) in this table that should reference
             the other table. For composite keys, supply a list of column names.
         ref_table
-            The reference table containing the valid values. Can be a DataFrame, database table, or
-            a callable that returns one (resolved at interrogation time).
+            The reference table containing the valid values. Can be a DataFrame, a database table,
+            a file path (e.g., to a CSV or Parquet file), a database connection string (e.g.,
+            `"duckdb:///path/to/db.ddb::table_name"`), or a callable that returns one of these. A
+            path or connection string is loaded at interrogation time, as with
+            [`tbl_match()`](`pointblank.Validate.tbl_match`); if the reference table can't be
+            loaded, the step is marked as having an evaluation error.
         ref_column
             The column (or list of columns for composite keys) in the reference table to check
             against. Must have the same length as ``columns``.
@@ -17265,8 +17275,7 @@ def interrogate(
                         ref_column = value["ref_column"]
                         in_table_columns = value["columns"]
 
-                        if callable(ref_table):
-                            ref_table = ref_table()
+                        ref_table = _load_ref_table(ref_table)
 
                         results_tbl = interrogate_in_table(
                             tbl=data_tbl_step,
@@ -25255,6 +25264,36 @@ def _transform_assertion_str(
     return type_upd
 
 
+def _load_ref_table(ref_table: Any) -> Any:
+    """
+    Resolve the `ref_table=` of `col_vals_in_table()` to a table at interrogation time.
+
+    A function is called to get the table, and a file path (CSV, Parquet, ...) or a database
+    connection string is loaded the way `tbl_match()` loads its comparison table. A reference that
+    can't be loaded, or isn't a table, raises a `StepEvaluationError`.
+    """
+    if callable(ref_table) and not hasattr(ref_table, "columns"):
+        ref_table = ref_table()
+
+    if isinstance(ref_table, (str, Path)):
+        try:
+            ref_table = _process_data(ref_table)
+        except Exception as e:
+            raise StepEvaluationError(
+                f"The reference table {str(ref_table)!r} couldn't be loaded ({_first_line(e)})."
+            ) from e
+
+    try:
+        _get_tbl_type(ref_table)
+    except TypeError as e:
+        raise StepEvaluationError(
+            "`ref_table=` must be a table, a file path, or a database connection string (or a "
+            f"function that returns one), not a {type(ref_table).__name__}."
+        ) from e
+
+    return ref_table
+
+
 def _apply_pre(pre: Callable, tbl: Any) -> Any:
     """
     Apply a `pre=` function to a table.

tests/test_col_vals_in_table.py

@@ -607,3 +607,64 @@ def test_failed_ref_conversion_is_eval_error():
     assert step.eval_error is True
     assert "connection lost" in step.notes["eval_error"]["text"]
     assert v.n_passed(i=2, scalar=True) == 2
+
+
+# ── Reference tables given as paths or connection strings ─────────────────────
+
+
+@pytest.fixture
+def ref_files(tmp_path):
+    """The reference table `x = [1, 2]` as a CSV file, a Parquet file, and a DuckDB table."""
+    import duckdb
+
+    ref = pl.DataFrame({"x": [1, 2]})
+    ref.write_csv(tmp_path / "ref.csv")
+    ref.write_parquet(tmp_path / "ref.parquet")
+    con = duckdb.connect(str(tmp_path / "ref.ddb"))
+    con.execute("CREATE TABLE refs AS SELECT * FROM (VALUES (1), (2)) AS t(x)")
+    con.close()
+    return tmp_path
+
+
+@pytest.mark.parametrize(
+    "make_ref",
+    [
+        lambda d: str(d / "ref.csv"),
+        lambda d: d / "ref.parquet",  # a pathlib.Path
+        lambda d: f"duckdb:///{d / 'ref.ddb'}::refs",
+        lambda d: lambda: str(d / "ref.csv"),  # a function returning a path
+    ],
+    ids=["csv", "parquet_path", "duckdb_connection", "callable"],
+)
+def test_ref_table_path_or_connection_string(ref_files, make_ref):
+    v = (
+        pb.Validate(data=pl.DataFrame({"a": [1, 2, 3]}))
+        .col_vals_in_table(columns="a", ref_table=make_ref(ref_files), ref_column="x")
+        .interrogate()
+    )
+
+    assert v.validation_info[0].eval_error is not True
+    assert v.n_passed(i=1, scalar=True) == 2
+    assert v.n_failed(i=1, scalar=True) == 1
+
+
+@pytest.mark.parametrize(
+    "ref_table, note",
+    [
+        ("does_not_exist.csv", "couldn't be loaded"),
+        (42, "must be a table, a file path, or a database connection string"),
+    ],
+)
+def test_ref_table_that_cannot_be_loaded_is_eval_error(ref_table, note):
+    """Before, these crashed `interrogate()` with a TypeError."""
+    v = (
+        pb.Validate(data=pl.DataFrame({"a": [1, 2, 3]}))
+        .col_vals_in_table(columns="a", ref_table=ref_table, ref_column="x")
+        .col_vals_not_null(columns="a")
+        .interrogate()
+    )
+
+    step = v.validation_info[0]
+    assert step.eval_error is True
+    assert note in step.notes["eval_error"]["text"]
+    assert v.n_passed(i=2, scalar=True) == 3

tests/test_plan_serialization.py

@@ -205,6 +205,31 @@ def test_to_code_col_vals_in_table_uses_ref_table_placeholder():
     assert "ref_table: your_reference_table" in yaml_text
 
 
+def test_col_vals_in_table_ref_path_roundtrips(tmp_path):
+    # A path (or connection string) is kept as-is, so the plan round-trips without a placeholder
+    import polars as pl
+
+    ref_path = tmp_path / "ref.csv"
+    pl.DataFrame({"x": [1, 2]}).write_csv(ref_path)
+    tbl = pl.DataFrame({"a": [1, 2, 3]})
+    validation = Validate(data=tbl).col_vals_in_table(
+        columns="a", ref_table=str(ref_path), ref_column="x"
+    )
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("error")
+        code = validation.to_code()
+        yaml_text = validation.to_yaml()
+
+    assert f"ref_table={str(ref_path)!r}" in code or f'ref_table="{ref_path}"' in code
+    assert _exec_code(code, tbl).interrogate().n_failed(i=1, scalar=True) == 1
+
+    # `set_tbl=` replaces the table, but the placeholder table name must still be loadable
+    yaml_text = yaml_text.replace("tbl: your_data", "tbl: small_table")
+    rebuilt = yaml_interrogate(yaml_text, set_tbl=tbl)
+    assert rebuilt.n_failed(i=1, scalar=True) == 1
+
+
 def test_to_yaml_roundtrips_via_yaml_interrogate(small_table):
     validation = (
         Validate(data=small_table, tbl_name="small_table", label="My Plan")