Merge pull request #225 from posit-dev/fix-centralize-data-ingest

fix: use `_process_data()` to centralize data ingest functionality

Author: rich-iannone

pointblank/assistant.py

@@ -180,21 +180,10 @@ def assistant(
     # If a dataset is provided, generate a table summary in JSON format
     if data is not None:
         # Import processing functions from validate module
-        from pointblank.validate import (
-            _process_connection_string,
-            _process_csv_input,
-            _process_parquet_input,
-        )
+        from pointblank.validate import _process_data
 
         # Process input data to handle different data source types
-        # Handle connection string input (e.g., "duckdb:///path/to/file.ddb::table_name")
-        data = _process_connection_string(data)
-
-        # Handle CSV file input (e.g., "data.csv" or Path("data.csv"))
-        data = _process_csv_input(data)
-
-        # Handle Parquet file input (e.g., "data.parquet", "data/*.parquet", "data/")
-        data = _process_parquet_input(data)
+        data = _process_data(data)
 
         scan = DataScan(data=data)
 

pointblank/cli.py

@@ -832,18 +832,12 @@ def preview(
 
             # If data has _row_num_ and it's not explicitly included, add it at the beginning
             try:
-                from pointblank.validate import (
-                    _process_connection_string,
-                    _process_csv_input,
-                    _process_parquet_input,
-                )
+                from pointblank.validate import _process_data
 
                 # Process the data source to get actual data object to check for _row_num_
                 processed_data = data
                 if isinstance(data, str):
-                    processed_data = _process_connection_string(data)
-                    processed_data = _process_csv_input(processed_data)
-                    processed_data = _process_parquet_input(processed_data)
+                    processed_data = _process_data(data)
 
                 # Get column names from the processed data
                 all_columns = []
@@ -861,18 +855,12 @@ def preview(
         elif col_range or col_first or col_last:
             # Need to get column names to apply range/first/last selection
             # Load the data to get column names
-            from pointblank.validate import (
-                _process_connection_string,
-                _process_csv_input,
-                _process_parquet_input,
-            )
+            from pointblank.validate import _process_data
 
             # Process the data source to get actual data object
             processed_data = data
             if isinstance(data, str):
-                processed_data = _process_connection_string(data)
-                processed_data = _process_csv_input(processed_data)
-                processed_data = _process_parquet_input(processed_data)
+                processed_data = _process_data(data)
 
             # Get column names from the processed data
             all_columns = []
@@ -935,17 +923,11 @@ def preview(
             # Get total dataset size before preview and gather metadata
             try:
                 # Process the data to get the actual data object for row count and metadata
-                from pointblank.validate import (
-                    _process_connection_string,
-                    _process_csv_input,
-                    _process_parquet_input,
-                )
+                from pointblank.validate import _process_data
 
                 processed_data = data
                 if isinstance(data, str):
-                    processed_data = _process_connection_string(data)
-                    processed_data = _process_csv_input(processed_data)
-                    processed_data = _process_parquet_input(processed_data)
+                    processed_data = _process_data(data)
 
                 total_dataset_rows = pb.get_row_count(processed_data)
 
@@ -1024,15 +1006,9 @@ def info(data_source: str):
                 source_type = f"External source: {data_source}"
 
                 # Process the data to get actual table object for inspection
-                from pointblank.validate import (
-                    _process_connection_string,
-                    _process_csv_input,
-                    _process_parquet_input,
-                )
+                from pointblank.validate import _process_data
 
-                data = _process_connection_string(data)
-                data = _process_csv_input(data)
-                data = _process_parquet_input(data)
+                data = _process_data(data)
                 console.print(f"[green]✓[/green] Loaded data source: {data_source}")
 
         # Get table information
@@ -1131,15 +1107,9 @@ def scan(
                     total_rows = None
             else:
                 # For file paths and connection strings, load the data first
-                from pointblank.validate import (
-                    _process_connection_string,
-                    _process_csv_input,
-                    _process_parquet_input,
-                )
+                from pointblank.validate import _process_data
 
-                processed_data = _process_connection_string(data)
-                processed_data = _process_csv_input(processed_data)
-                processed_data = _process_parquet_input(processed_data)
+                processed_data = _process_data(data)
                 scan_result = pb.col_summary_tbl(data=processed_data)
                 source_type = f"External source: {data_source}"
                 table_type = _get_tbl_type(processed_data)
@@ -1212,16 +1182,10 @@ def missing(data_source: str, output_html: str | None):
             original_data = data
             if isinstance(data, str):
                 # Process the data to get the actual data object
-                from pointblank.validate import (
-                    _process_connection_string,
-                    _process_csv_input,
-                    _process_parquet_input,
-                )
+                from pointblank.validate import _process_data
 
                 try:
-                    original_data = _process_connection_string(data)
-                    original_data = _process_csv_input(original_data)
-                    original_data = _process_parquet_input(original_data)
+                    original_data = _process_data(data)
                 except Exception:  # pragma: no cover
                     pass  # Use the string data as fallback
 

pointblank/compare.py

@@ -11,21 +11,13 @@
 class Compare:
     def __init__(self, a: IntoFrame, b: IntoFrame) -> None:
         # Import processing functions from validate module
-        from pointblank.validate import (
-            _process_connection_string,
-            _process_csv_input,
-            _process_parquet_input,
-        )
+        from pointblank.validate import _process_data
 
         # Process input data for table a
-        a = _process_connection_string(a)
-        a = _process_csv_input(a)
-        a = _process_parquet_input(a)
+        a = _process_data(a)
 
         # Process input data for table b
-        b = _process_connection_string(b)
-        b = _process_csv_input(b)
-        b = _process_parquet_input(b)
+        b = _process_data(b)
 
         self.a: IntoFrame = a
         self.b: IntoFrame = b

pointblank/datascan.py

@@ -126,24 +126,11 @@ class DataScan:
     def __init__(self, data: IntoFrameT, tbl_name: str | None = None) -> None:
         # Import processing functions from validate module
         from pointblank.validate import (
-            _process_connection_string,
-            _process_csv_input,
-            _process_github_url,
-            _process_parquet_input,
+            _process_data,
         )
 
         # Process input data to handle different data source types
-        # Handle GitHub URL input (e.g., "https://github.com/user/repo/blob/main/data.csv")
-        data = _process_github_url(data)
-
-        # Handle connection string input (e.g., "duckdb:///path/to/file.ddb::table_name")
-        data = _process_connection_string(data)
-
-        # Handle CSV file input (e.g., "data.csv" or Path("data.csv"))
-        data = _process_csv_input(data)
-
-        # Handle Parquet file input (e.g., "data.parquet", "data/*.parquet", "data/")
-        data = _process_parquet_input(data)
+        data = _process_data(data)
 
         as_native = nw.from_native(data)
 
@@ -596,25 +583,10 @@ def col_summary_tbl(data: FrameT | Any, tbl_name: str | None = None) -> GT:
     """
 
     # Import processing functions from validate module
-    from pointblank.validate import (
-        _process_connection_string,
-        _process_csv_input,
-        _process_github_url,
-        _process_parquet_input,
-    )
+    from pointblank.validate import _process_data
 
     # Process input data to handle different data source types
-    # Handle GitHub URL input (e.g., "https://github.com/user/repo/blob/main/data.csv")
-    data = _process_github_url(data)
-
-    # Handle connection string input (e.g., "duckdb:///path/to/file.ddb::table_name")
-    data = _process_connection_string(data)
-
-    # Handle CSV file input (e.g., "data.csv" or Path("data.csv"))
-    data = _process_csv_input(data)
-
-    # Handle Parquet file input (e.g., "data.parquet", "data/*.parquet", "data/")
-    data = _process_parquet_input(data)
+    data = _process_data(data)
 
     scanner = DataScan(data=data, tbl_name=tbl_name)
     return scanner.get_tabular_report()

pointblank/validate.py

@@ -735,9 +735,51 @@ def get_data_path(
                 return tmp_file.name
 
 
-# =============================================================================
-# Utility functions for processing input data (shared by preview() and Validate class)
-# =============================================================================
+def _process_data(data: FrameT | Any) -> FrameT | Any:
+    """
+    Centralized data processing pipeline that handles all supported input types.
+
+    This function consolidates the data processing pipeline used across multiple
+    classes and functions in Pointblank. It processes data through a consistent
+    sequence of transformations to handle different data source types.
+
+    The processing order is important:
+
+    1. GitHub URLs (must come before connection string processing)
+    2. Database connection strings
+    3. CSV file paths
+    4. Parquet file paths
+
+    Parameters
+    ----------
+    data : FrameT | Any
+        The input data which could be:
+        - a DataFrame object (Polars, Pandas, Ibis, etc.)
+        - a GitHub URL pointing to a CSV or Parquet file
+        - a database connection string (e.g., "duckdb:///path/to/file.ddb::table_name")
+        - a CSV file path (string or Path object with .csv extension)
+        - a Parquet file path, glob pattern, directory, or partitioned dataset
+        - any other data type (returned unchanged)
+
+    Returns
+    -------
+    FrameT | Any
+        Processed data as a DataFrame if input was a supported data source type,
+        otherwise the original data unchanged.
+    """
+    # Handle GitHub URL input (e.g., "https://github.com/user/repo/blob/main/data.csv")
+    data = _process_github_url(data)
+
+    # Handle connection string input (e.g., "duckdb:///path/to/file.ddb::table_name")
+    data = _process_connection_string(data)
+
+    # Handle CSV file input (e.g., "data.csv" or Path("data.csv"))
+    data = _process_csv_input(data)
+
+    # Handle Parquet file input (e.g., "data.parquet", "data/*.parquet", "data/")
+    data = _process_parquet_input(data)
+
+    return data
 
 
 def _process_github_url(data: FrameT | Any) -> FrameT | Any:
@@ -1321,17 +1363,7 @@ def preview(
     """
 
     # Process input data to handle different data source types
-    # Handle GitHub URL input (e.g., "https://github.com/user/repo/blob/main/data.csv")
-    data = _process_github_url(data)
-
-    # Handle connection string input (e.g., "duckdb:///path/to/file.ddb::table_name")
-    data = _process_connection_string(data)
-
-    # Handle CSV file input (e.g., "data.csv" or Path("data.csv"))
-    data = _process_csv_input(data)
-
-    # Handle Parquet file input (e.g., "data.parquet", "data/*.parquet", "data/")
-    data = _process_parquet_input(data)
+    data = _process_data(data)
 
     if incl_header is None:
         incl_header = global_config.preview_incl_header
@@ -1816,17 +1848,7 @@ def missing_vals_tbl(data: FrameT | Any) -> GT:
     """
 
     # Process input data to handle different data source types
-    # Handle GitHub URL input (e.g., "https://github.com/user/repo/blob/main/data.csv")
-    data = _process_github_url(data)
-
-    # Handle connection string input (e.g., "duckdb:///path/to/file.ddb::table_name")
-    data = _process_connection_string(data)
-
-    # Handle CSV file input (e.g., "data.csv" or Path("data.csv"))
-    data = _process_csv_input(data)
-
-    # Handle Parquet file input (e.g., "data.parquet", "data/*.parquet", "data/")
-    data = _process_parquet_input(data)
+    data = _process_data(data)
 
     # Make a copy of the data to avoid modifying the original
     data = copy.deepcopy(data)
@@ -2431,14 +2453,7 @@ def get_column_count(data: FrameT | Any) -> int:
 
     # Process different input types
     if isinstance(data, str) or isinstance(data, Path):
-        # Process GitHub URLs first
-        data = _process_github_url(data)
-        # Handle connection string input
-        data = _process_connection_string(data)
-        # Handle CSV file input
-        data = _process_csv_input(data)
-        # Handle Parquet file input
-        data = _process_parquet_input(data)
+        data = _process_data(data)
     elif isinstance(data, list):
         # Handle list of file paths (likely Parquet files)
         data = _process_parquet_input(data)
@@ -2607,14 +2622,7 @@ def get_row_count(data: FrameT | Any) -> int:
 
     # Process different input types
     if isinstance(data, str) or isinstance(data, Path):
-        # Process GitHub URLs first
-        data = _process_github_url(data)
-        # Handle connection string input
-        data = _process_connection_string(data)
-        # Handle CSV file input
-        data = _process_csv_input(data)
-        # Handle Parquet file input
-        data = _process_parquet_input(data)
+        data = _process_data(data)
     elif isinstance(data, list):
         # Handle list of file paths (likely Parquet files)
         data = _process_parquet_input(data)
@@ -3550,17 +3558,8 @@ def send_report():
     locale: str | None = None
 
     def __post_init__(self):
-        # Handle GitHub URL input for the data parameter
-        self.data = _process_github_url(self.data)
-
-        # Handle connection string input for the data parameter
-        self.data = _process_connection_string(self.data)
-
-        # Handle CSV file input for the data parameter
-        self.data = _process_csv_input(self.data)
-
-        # Handle Parquet file input for the data parameter
-        self.data = _process_parquet_input(self.data)
+        # Process data through the centralized data processing pipeline
+        self.data = _process_data(self.data)
 
         # Check input of the `thresholds=` argument
         _check_thresholds(thresholds=self.thresholds)