API Reference

The building blocks of Veridelta. Explore the configuration schemas, core diffing engine, and operational utilities.

Configuration Models

Strict Pydantic models defining how Veridelta comparisons are structured. These can be instantiated programmatically or driven by declarative YAML.

Data models for Veridelta configuration and results.

This module defines the Pydantic models used to configure data ingestion, comparison rules, and format the output summaries. It acts as the strict schema definition for the YAML configuration files.

`SchemaMode = Literal['exact', 'allow_additions', 'allow_removals', 'intersection']` `module-attribute`

Defines how strictly the engine enforces column schemas between datasets.

"exact": Strict 1:1 mapping. Columns must be identical and in the exact same order.
"allow_additions": Target can have new columns, but must contain every column present in the Source.
"allow_removals": Target is allowed to drop legacy columns, but cannot add any new columns.
"intersection": Only diff columns that exist in both datasets, ignoring all others. (Default)

`SourceType = Literal['csv', 'json', 'parquet', 'fixed_width', 'netcdf', 'shapefile', 'geopackage', 'excel', 'sql', 'delta', 'avro', 'xml', 'arrow']` `module-attribute`

Supported and roadmap data formats for ingestion.

`WhitespaceMode = Literal['none', 'left', 'right', 'both']` `module-attribute`

Granular control over string whitespace stripping.

"none": Do not strip any whitespace.
"left": Strip leading whitespace only.
"right": Strip trailing whitespace only.
"both": Strip both leading and trailing whitespace.

`DiffConfig`

Bases: BaseModel

The master configuration for a Veridelta comparison run.

Attributes:

Name	Type	Description
`primary_keys`	`list[str]`	Columns used to join and align the datasets. Must be unique in both datasets.
`schema_mode`	`SchemaMode`	How strictly to enforce column existence and matching between sources.
`strict_types`	`bool`	If False (default), the engine implicitly soft-casts target columns to source types purely for the comparison expression, preventing execution crashes on type mismatches. If True, type mismatches will automatically evaluate as row failures.
`normalize_column_names`	`bool`	If True, strips whitespace and lowercases all column headers prior to schema alignment.
`default_absolute_tolerance`	`float`	Global absolute tolerance for numeric columns.
`default_relative_tolerance`	`float`	Global relative tolerance for numeric columns.
`default_treat_null_as_equal`	`bool`	Global setting for handling NULL == NULL.
`default_whitespace_mode`	`WhitespaceMode`	Global string whitespace stripping mode.
`default_null_values`	`list[str]`	Global list of string values to aggressively coerce to NULL.
`rules`	`list[DiffRule]`	List of per-column comparison overrides. Specific `column_names` take precedence over regex `pattern` rules.
`threshold`	`float`	Allowed mismatch ratio (0.0 to 1.0) before the `is_match` flag evaluates to False.
`report_top_columns_limit`	`int`	Max number of drifted columns to display in the generated markdown report summary.
`output_path`	`str \| None`	Optional path to save the resulting diff report and artifacts (added, removed, and changed rows).
`output_format`	`str`	The file format for exported discrepancy artifacts (e.g., 'parquet', 'csv').

Source code in src/veridelta/models.py

class DiffConfig(BaseModel):
    """The master configuration for a Veridelta comparison run.

    Attributes:
        primary_keys (list[str]): Columns used to join and align the datasets.
            Must be unique in both datasets.
        schema_mode (SchemaMode): How strictly to enforce column existence and
            matching between sources.
        strict_types (bool): If False (default), the engine implicitly soft-casts
            target columns to source types purely for the comparison expression,
            preventing execution crashes on type mismatches. If True, type mismatches
            will automatically evaluate as row failures.
        normalize_column_names (bool): If True, strips whitespace and lowercases
            all column headers prior to schema alignment.
        default_absolute_tolerance (float): Global absolute tolerance for numeric columns.
        default_relative_tolerance (float): Global relative tolerance for numeric columns.
        default_treat_null_as_equal (bool): Global setting for handling NULL == NULL.
        default_whitespace_mode (WhitespaceMode): Global string whitespace stripping mode.
        default_null_values (list[str]): Global list of string values to aggressively
            coerce to NULL.
        rules (list[DiffRule]): List of per-column comparison overrides. Specific
            `column_names` take precedence over regex `pattern` rules.
        threshold (float): Allowed mismatch ratio (0.0 to 1.0) before the `is_match`
            flag evaluates to False.
        report_top_columns_limit (int): Max number of drifted columns to display
            in the generated markdown report summary.
        output_path (str | None): Optional path to save the resulting diff report
            and artifacts (added, removed, and changed rows).
        output_format (str): The file format for exported discrepancy artifacts
            (e.g., 'parquet', 'csv').
    """

    model_config = ConfigDict(extra="forbid")

    primary_keys: list[str] = Field(..., description="Columns used to join datasets.")

    schema_mode: SchemaMode = Field(
        default="intersection",
        description="Schema enforcement mode: 'exact', 'allow_additions', 'allow_removals', or 'intersection'.",
    )
    strict_types: bool = Field(
        default=False,
        description="If False, engine attempts to safely cast Target columns to Source types.",
    )

    normalize_column_names: bool = Field(
        default=False,
        description="If True, strips whitespace and lowercases all column headers before processing.",
    )

    default_absolute_tolerance: float = Field(
        default=0.0, ge=0.0, description="Global absolute tolerance for numeric columns."
    )
    default_relative_tolerance: float = Field(
        default=0.0, ge=0.0, description="Global relative tolerance for numeric columns."
    )
    default_treat_null_as_equal: bool = Field(
        default=True, description="Globally treat NULL == NULL as a match."
    )
    default_whitespace_mode: WhitespaceMode = Field(
        default="none",
        description="Global string whitespace stripping mode: 'none', 'left', 'right', or 'both'.",
    )
    default_null_values: list[str] = Field(
        default_factory=list, description="Global list of string values to coerce to NULL."
    )

    rules: list[DiffRule] = Field(default_factory=list, description="Column overrides.")

    threshold: float = Field(
        default=0.0, ge=0.0, le=1.0, description="Allowed mismatch percentage (0.0 to 1.0)."
    )

    report_top_columns_limit: int = Field(
        default=5,
        ge=0,
        description="Max number of top drifted columns to show in the report summary.",
    )

    output_path: str | None = Field(
        default=None, description="Optional path to save the detailed diff report."
    )
    output_format: str = Field(
        default="parquet",
        description="The file format for exported discrepancy artifacts (e.g., 'parquet', 'csv').",
    )

    @model_validator(mode="after")
    def apply_schema_normalization(self) -> "DiffConfig":
        """Automatically lowercases and strips config keys if normalization is enabled.

        Returns:
            DiffConfig: The mutated configuration instance.
        """
        if self.normalize_column_names:
            self.primary_keys = [pk.strip().lower() for pk in self.primary_keys]

            rules: list[DiffRule] = self.rules
            for rule in rules:
                rule.column_names = [col.strip().lower() for col in rule.column_names]

        return self

`apply_schema_normalization()`

Automatically lowercases and strips config keys if normalization is enabled.

Returns:

Name	Type	Description
`DiffConfig`	`DiffConfig`	The mutated configuration instance.

Source code in src/veridelta/models.py

@model_validator(mode="after")
def apply_schema_normalization(self) -> "DiffConfig":
    """Automatically lowercases and strips config keys if normalization is enabled.

    Returns:
        DiffConfig: The mutated configuration instance.
    """
    if self.normalize_column_names:
        self.primary_keys = [pk.strip().lower() for pk in self.primary_keys]

        rules: list[DiffRule] = self.rules
        for rule in rules:
            rule.column_names = [col.strip().lower() for col in rule.column_names]

    return self

`DiffRule`

Bases: BaseModel

Specific overrides for one or more columns using exact names or regex.

Attributes:

Name	Type	Description
`column_names`	`list[str]`	Exact names of the columns in the source dataset.
`pattern`	`str \| None`	Regex pattern to match multiple columns (e.g., '^AMT_.*').
`absolute_tolerance`	`float \| None`	The maximum allowed absolute difference for numeric mathematical comparisons.
`relative_tolerance`	`float \| None`	The maximum allowed relative difference (e.g., 0.01 for 1%).
`case_insensitive`	`bool \| None`	If True, ignores case differences in strings.
`whitespace_mode`	`WhitespaceMode \| None`	Granular control over stripping leading/trailing whitespace prior to string comparison.
`regex_replace`	`dict[str, str] \| None`	Dictionary of `{pattern: replacement}` to sanitize text. Implicitly executed before type coercion to ensure text sanitization completes safely before casting.
`pad_zeros`	`int \| None`	Left-pad numeric strings to this exact length (e.g., 5 -> '00123').
`value_map`	`dict[str, str] \| None`	Translate Source values to Target values before comparison (e.g., `{'M': 'Male'}`).
`null_values`	`list[str] \| None`	Specific string values to actively coerce to NULL (e.g., `['N/A', '-999']`).
`treat_null_as_equal`	`bool \| None`	If True, evaluates NULL == NULL as a successful match rather than a missing value mismatch.
`datetime_format`	`str \| None`	Expected strptime format for dates (e.g., '%Y-%m-%d %H:%M:%S').
`timezone`	`str \| None`	Target timezone to normalize dates to before comparison.
`cast_to`	`str \| None`	Explicitly cast column to this Polars datatype (e.g., 'Float64'). Evaluated after string transformations.
`ignore`	`bool`	Whether to skip this column entirely during comparison.
`rename_to`	`str \| None`	The name in the target dataset if it differs from the source. Only valid when `column_names` contains exactly one entry.

Source code in src/veridelta/models.py

class DiffRule(BaseModel):
    """Specific overrides for one or more columns using exact names or regex.

    Attributes:
        column_names (list[str]): Exact names of the columns in the source dataset.
        pattern (str | None): Regex pattern to match multiple columns (e.g., '^AMT_.*').
        absolute_tolerance (float | None): The maximum allowed absolute difference
            for numeric mathematical comparisons.
        relative_tolerance (float | None): The maximum allowed relative difference
            (e.g., 0.01 for 1%).
        case_insensitive (bool | None): If True, ignores case differences in strings.
        whitespace_mode (WhitespaceMode | None): Granular control over stripping
            leading/trailing whitespace prior to string comparison.
        regex_replace (dict[str, str] | None): Dictionary of `{pattern: replacement}`
            to sanitize text. Implicitly executed *before* type coercion to ensure
            text sanitization completes safely before casting.
        pad_zeros (int | None): Left-pad numeric strings to this exact length
            (e.g., 5 -> '00123').
        value_map (dict[str, str] | None): Translate Source values to Target values
            before comparison (e.g., `{'M': 'Male'}`).
        null_values (list[str] | None): Specific string values to actively coerce
            to NULL (e.g., `['N/A', '-999']`).
        treat_null_as_equal (bool | None): If True, evaluates NULL == NULL as a
            successful match rather than a missing value mismatch.
        datetime_format (str | None): Expected strptime format for dates
            (e.g., '%Y-%m-%d %H:%M:%S').
        timezone (str | None): Target timezone to normalize dates to before comparison.
        cast_to (str | None): Explicitly cast column to this Polars datatype
            (e.g., 'Float64'). Evaluated *after* string transformations.
        ignore (bool): Whether to skip this column entirely during comparison.
        rename_to (str | None): The name in the target dataset if it differs from
            the source. Only valid when `column_names` contains exactly one entry.
    """

    model_config = ConfigDict(extra="forbid")

    column_names: list[str] = Field(
        default_factory=list, description="Exact names of the columns in the source."
    )
    pattern: str | None = Field(
        default=None, description="Regex pattern to match multiple columns (e.g., '^AMT_.*')."
    )

    absolute_tolerance: float | None = Field(
        default=None, ge=0.0, description="Absolute tolerance for numeric differences."
    )
    relative_tolerance: float | None = Field(
        default=None, ge=0.0, description="Relative tolerance (e.g., 0.01 for 1%)."
    )

    case_insensitive: bool | None = Field(
        default=None, description="Ignore case for string comparisons."
    )
    whitespace_mode: WhitespaceMode | None = Field(
        default=None,
        description="Whitespace stripping mode: 'none', 'left', 'right', or 'both'.",
    )
    regex_replace: dict[str, str] | None = Field(
        default=None,
        description="Dictionary of {regex_pattern: replacement_string} to sanitize text.",
    )
    pad_zeros: int | None = Field(
        default=None,
        ge=0,
        description="Left-pad numeric strings to this length (e.g., 5 -> '00123').",
    )

    value_map: dict[str, str] | None = Field(
        default=None, description="Translate Source values to Target values (e.g., {'M': 'Male'})."
    )
    null_values: list[str] | None = Field(
        default=None, description="Specific string values to treat as NULL (e.g., ['N/A', '-999'])."
    )
    treat_null_as_equal: bool | None = Field(
        default=None, description="Treat missing values (NULL/None) in both sources as a match."
    )

    datetime_format: str | None = Field(
        default=None, description="Expected strptime format (e.g., '%Y-%m-%d %H:%M:%S')."
    )
    timezone: str | None = Field(
        default=None, description="Target timezone to normalize dates to before comparison."
    )

    cast_to: str | None = Field(
        default=None,
        description="Explicitly cast column to this Polars datatype (e.g., 'Float64').",
    )
    ignore: bool = Field(
        default=False, description="If True, this column will be excluded from the comparison."
    )
    rename_to: str | None = Field(
        default=None,
        description="Name in target dataset if different (use only for single columns).",
    )

    @field_validator("pattern")
    @classmethod
    def validate_pattern(cls, v: str | None) -> str | None:
        """Ensures the provided regex pattern is a valid expression at configuration time.

        Args:
            v (str | None): The string regex pattern to validate.

        Returns:
            str | None: The validated regex string.

        Raises:
            ValueError: If the regex pattern cannot be compiled.
        """
        if v is not None:
            try:
                re.compile(v)
            except re.error as err:
                raise ValueError(f"Invalid regex pattern '{v}': {err}") from err
        return v

    @field_validator("regex_replace")
    @classmethod
    def validate_regex_replace(cls, v: dict[str, str] | None) -> dict[str, str] | None:
        """Ensures all keys in the regex replacement dictionary are valid regex patterns.

        Args:
            v (dict[str, str] | None): A mapping of regex patterns to replacements.

        Returns:
            dict[str, str] | None: The validated dictionary.

        Raises:
            ValueError: If any key in the dictionary is an invalid regex pattern.
        """
        if v is not None:
            for pattern in v:
                try:
                    re.compile(pattern)
                except re.error as err:  # noqa: PERF203
                    raise ValueError(f"Invalid regex replace pattern '{pattern}': {err}") from err
        return v

`validate_pattern(v)` `classmethod`

Ensures the provided regex pattern is a valid expression at configuration time.

Parameters:

Name	Type	Description	Default
`v`	`str \| None`	The string regex pattern to validate.	required

Returns:

Type	Description
`str \| None`	str \| None: The validated regex string.

Raises:

Type	Description
`ValueError`	If the regex pattern cannot be compiled.

Source code in src/veridelta/models.py

@field_validator("pattern")
@classmethod
def validate_pattern(cls, v: str | None) -> str | None:
    """Ensures the provided regex pattern is a valid expression at configuration time.

    Args:
        v (str | None): The string regex pattern to validate.

    Returns:
        str | None: The validated regex string.

    Raises:
        ValueError: If the regex pattern cannot be compiled.
    """
    if v is not None:
        try:
            re.compile(v)
        except re.error as err:
            raise ValueError(f"Invalid regex pattern '{v}': {err}") from err
    return v

`validate_regex_replace(v)` `classmethod`

Ensures all keys in the regex replacement dictionary are valid regex patterns.

Parameters:

Name	Type	Description	Default
`v`	`dict[str, str] \| None`	A mapping of regex patterns to replacements.	required

Returns:

Type	Description
`dict[str, str] \| None`	dict[str, str] \| None: The validated dictionary.

Raises:

Type	Description
`ValueError`	If any key in the dictionary is an invalid regex pattern.

Source code in src/veridelta/models.py

@field_validator("regex_replace")
@classmethod
def validate_regex_replace(cls, v: dict[str, str] | None) -> dict[str, str] | None:
    """Ensures all keys in the regex replacement dictionary are valid regex patterns.

    Args:
        v (dict[str, str] | None): A mapping of regex patterns to replacements.

    Returns:
        dict[str, str] | None: The validated dictionary.

    Raises:
        ValueError: If any key in the dictionary is an invalid regex pattern.
    """
    if v is not None:
        for pattern in v:
            try:
                re.compile(pattern)
            except re.error as err:  # noqa: PERF203
                raise ValueError(f"Invalid regex replace pattern '{pattern}': {err}") from err
    return v

`DiffSummary`

Bases: BaseModel

The high-level execution results of a Veridelta comparison.

Attributes:

Name	Type	Description
`total_rows_source`	`int`	Number of rows in the source dataset.
`total_rows_target`	`int`	Number of rows in the target dataset.
`added_count`	`int`	Rows found only in the target (missing from source).
`removed_count`	`int`	Rows found only in the source (missing from target).
`changed_count`	`int`	Rows present in both datasets but with value differences.
`column_mismatches`	`dict[str, int]`	Dictionary mapping column names to the exact count of mismatched rows for that specific column.
`is_match`	`bool`	Boolean indicating if the overall diff falls within the allowed mismatch threshold.
`total_mismatches`	`int`	(Computed) The sum of all added, removed, and changed rows.
`mismatch_ratio`	`float`	(Computed) The ratio of mismatched rows to the baseline source dataset.
`match_rate_percentage`	`float`	(Computed) The overall match rate expressed as a percentage (e.g., 99.98).
`is_perfect_match`	`bool`	(Computed) True only if there are exactly 0 mismatches.
`volume_shift`	`int`	(Computed) The net change in row volume (Target - Source).
`report_summary`	`str`	(Computed) A pre-formatted, human-readable markdown status report intended for CI/CD logs or PR comments.
`report_limit`	`int`	Internal configuration dictating the max columns to display in the `report_summary`. Implicitly excluded from JSON serialization.

Source code in src/veridelta/models.py

class DiffSummary(BaseModel):
    """The high-level execution results of a Veridelta comparison.

    Attributes:
        total_rows_source (int): Number of rows in the source dataset.
        total_rows_target (int): Number of rows in the target dataset.
        added_count (int): Rows found only in the target (missing from source).
        removed_count (int): Rows found only in the source (missing from target).
        changed_count (int): Rows present in both datasets but with value differences.
        column_mismatches (dict[str, int]): Dictionary mapping column names to the
            exact count of mismatched rows for that specific column.
        is_match (bool): Boolean indicating if the overall diff falls within the
            allowed mismatch threshold.
        total_mismatches (int): (Computed) The sum of all added, removed, and changed rows.
        mismatch_ratio (float): (Computed) The ratio of mismatched rows to the baseline
            source dataset.
        match_rate_percentage (float): (Computed) The overall match rate expressed
            as a percentage (e.g., 99.98).
        is_perfect_match (bool): (Computed) True only if there are exactly 0 mismatches.
        volume_shift (int): (Computed) The net change in row volume (Target - Source).
        report_summary (str): (Computed) A pre-formatted, human-readable markdown
            status report intended for CI/CD logs or PR comments.
        report_limit (int): Internal configuration dictating the max columns to display
            in the `report_summary`. Implicitly excluded from JSON serialization.
    """

    model_config = ConfigDict(extra="forbid")

    total_rows_source: int
    total_rows_target: int
    added_count: int
    removed_count: int
    changed_count: int
    column_mismatches: dict[str, int] = Field(default_factory=dict)
    is_match: bool

    report_limit: int = Field(default=5, exclude=True)

    @computed_field
    @property
    def total_mismatches(self) -> int:
        """Calculates the sum of all added, removed, and changed rows.

        Returns:
            int: The total count of discrepancy events.
        """
        return self.added_count + self.removed_count + self.changed_count

    @computed_field
    @property
    def mismatch_ratio(self) -> float:
        """Calculates the ratio of mismatched rows to the baseline source dataset.

        Returns:
            float: A float representing the ratio (0.0 to 1.0+).
        """
        return float(self.total_mismatches) / float(max(self.total_rows_source, 1))

    @computed_field
    @property
    def match_rate_percentage(self) -> float:
        """Calculates the overall match rate expressed as a percentage.

        Returns:
            float: The match percentage rounded to two decimal places (e.g., 99.98).
        """
        return round((1.0 - self.mismatch_ratio) * 100.0, 2)

    @computed_field
    @property
    def is_perfect_match(self) -> bool:
        """Evaluates if the datasets are completely identical under the configured rules.

        Returns:
            bool: True if there are zero total mismatches.
        """
        return self.total_mismatches == 0

    @computed_field
    @property
    def volume_shift(self) -> int:
        """Calculates the net change in row volume between the systems.

        Returns:
            int: The net shift (Target rows - Source rows).
        """
        return self.total_rows_target - self.total_rows_source

    @computed_field
    @property
    def report_summary(self) -> str:
        """Generates a pre-formatted, human-readable status report.

        This string aggregates all metrics and top column-level drifts into a
        clean markdown format ready for immediate pipeline logging.

        Returns:
            str: The formatted execution summary.
        """
        status_icon = "PASSED" if self.is_match else "FAILED"
        perfect_tag = " (Perfect Match)" if self.is_perfect_match else ""

        base_report = (
            f"Veridelta Execution Summary\n"
            f"===========================\n"
            f"Status:        {status_icon}{perfect_tag}\n"
            f"Match Rate:    {self.match_rate_percentage}%\n"
            f"Source Rows:   {self.total_rows_source:,}\n"
            f"Target Rows:   {self.total_rows_target:,}\n"
            f"Volume Shift:  {self.volume_shift:+,} rows\n"
            f"\nRow-Level Discrepancies:\n"
            f"---------------------------\n"
            f"Added:         {self.added_count:,}\n"
            f"Removed:       {self.removed_count:,}\n"
            f"Changed:       {self.changed_count:,}\n"
            f"Total Issues:  {self.total_mismatches:,}\n"
        )

        if not self.column_mismatches or self.report_limit == 0:
            return base_report

        top_cols = sorted(self.column_mismatches.items(), key=lambda x: x[1], reverse=True)[
            : self.report_limit
        ]

        col_report = "\nTop Column-Level Drifts:\n---------------------------\n"
        for col, count in top_cols:
            col_report += f"- {col}: {count:,} mismatches\n"

        return base_report + col_report

`is_perfect_match` `property`

Evaluates if the datasets are completely identical under the configured rules.

Returns:

Name	Type	Description
`bool`	`bool`	True if there are zero total mismatches.

`match_rate_percentage` `property`

Calculates the overall match rate expressed as a percentage.

Returns:

Name	Type	Description
`float`	`float`	The match percentage rounded to two decimal places (e.g., 99.98).

`mismatch_ratio` `property`

Calculates the ratio of mismatched rows to the baseline source dataset.

Returns:

Name	Type	Description
`float`	`float`	A float representing the ratio (0.0 to 1.0+).

`report_summary` `property`

Generates a pre-formatted, human-readable status report.

This string aggregates all metrics and top column-level drifts into a clean markdown format ready for immediate pipeline logging.

Returns:

Name	Type	Description
`str`	`str`	The formatted execution summary.

`total_mismatches` `property`

Calculates the sum of all added, removed, and changed rows.

Returns:

Name	Type	Description
`int`	`int`	The total count of discrepancy events.

`volume_shift` `property`

Calculates the net change in row volume between the systems.

Returns:

Name	Type	Description
`int`	`int`	The net shift (Target rows - Source rows).

`SourceConfig`

Bases: BaseModel

Configuration for a specific data source.

Attributes:

Name	Type	Description
`path`	`str`	File system path or URI to the data.
`format`	`SourceType`	The format of the file (e.g., 'csv', 'parquet').
`options`	`dict[str, Any]`	Format-specific keyword arguments passed directly to the underlying Polars reader (e.g., `{'separator': ';'}`).

Source code in src/veridelta/models.py

class SourceConfig(BaseModel):
    """Configuration for a specific data source.

    Attributes:
        path (str): File system path or URI to the data.
        format (SourceType): The format of the file (e.g., 'csv', 'parquet').
        options (dict[str, Any]): Format-specific keyword arguments passed
            directly to the underlying Polars reader (e.g., `{'separator': ';'}`).
    """

    model_config = ConfigDict(extra="forbid")

    path: str = Field(..., description="File system path or URI to the data.")
    format: SourceType = Field("csv", description="The format of the file.")
    options: dict[str, Any] = Field(
        default_factory=dict,
        description="Format-specific options (e.g., {'separator': ';'}).",
    )

The Engine

The core mathematical evaluation engine and I/O orchestration, powered by the Rust-based Polars backend.

Core engine for data ingestion and alignment.

This module houses the I/O loaders, the DataIngestor for dataset preparation, and the DiffEngine which performs the high-performance Polars comparisons.

`BaseLoader`

Bases: ABC

Abstract base class for all data loaders.

Source code in src/veridelta/engine.py

class BaseLoader(ABC):
    """Abstract base class for all data loaders."""

    @abstractmethod
    def load(self, config: SourceConfig) -> pl.LazyFrame:
        """Loads data from a source into a Polars LazyFrame.

        Args:
            config (SourceConfig): The configuration detailing the path, format,
                and format-specific parsing options.

        Returns:
            pl.LazyFrame: The lazy-loaded dataset graph.
        """
        pass

`load(config)` `abstractmethod`

Loads data from a source into a Polars LazyFrame.

Parameters:

Name	Type	Description	Default
`config`	`SourceConfig`	The configuration detailing the path, format, and format-specific parsing options.	required

Returns:

Type	Description
`LazyFrame`	pl.LazyFrame: The lazy-loaded dataset graph.

Source code in src/veridelta/engine.py

@abstractmethod
def load(self, config: SourceConfig) -> pl.LazyFrame:
    """Loads data from a source into a Polars LazyFrame.

    Args:
        config (SourceConfig): The configuration detailing the path, format,
            and format-specific parsing options.

    Returns:
        pl.LazyFrame: The lazy-loaded dataset graph.
    """
    pass

`CSVLoader`

Bases: BaseLoader

Loader for CSV files utilizing the fast Polars CSV scanner.

Source code in src/veridelta/engine.py

class CSVLoader(BaseLoader):
    """Loader for CSV files utilizing the fast Polars CSV scanner."""

    def load(self, config: SourceConfig) -> pl.LazyFrame:
        """Loads a CSV file into a Polars LazyFrame.

        Args:
            config (SourceConfig): The source configuration. Extra options are
                passed directly to `pl.scan_csv`.

        Returns:
            pl.LazyFrame: The lazy dataset graph.
        """
        return pl.scan_csv(config.path, **config.options)

`load(config)`

Loads a CSV file into a Polars LazyFrame.

Parameters:

Name	Type	Description	Default
`config`	`SourceConfig`	The source configuration. Extra options are passed directly to `pl.scan_csv`.	required

Returns:

Type	Description
`LazyFrame`	pl.LazyFrame: The lazy dataset graph.

Source code in src/veridelta/engine.py

def load(self, config: SourceConfig) -> pl.LazyFrame:
    """Loads a CSV file into a Polars LazyFrame.

    Args:
        config (SourceConfig): The source configuration. Extra options are
            passed directly to `pl.scan_csv`.

    Returns:
        pl.LazyFrame: The lazy dataset graph.
    """
    return pl.scan_csv(config.path, **config.options)

`DataIngestor`

Coordinates the loading, renaming, and structural alignment of datasets.

This class prepares raw external data for comparison by normalizing headers and dropping ignored columns before handing them off to the DiffEngine.

Source code in src/veridelta/engine.py

class DataIngestor:
    """Coordinates the loading, renaming, and structural alignment of datasets.

    This class prepares raw external data for comparison by normalizing headers
    and dropping ignored columns before handing them off to the DiffEngine.
    """

    def __init__(
        self, diff_config: DiffConfig, source_config: SourceConfig, target_config: SourceConfig
    ) -> None:
        """Initializes the ingestor.

        Args:
            diff_config (DiffConfig): The master comparison configuration.
            source_config (SourceConfig): File and format settings for the source.
            target_config (SourceConfig): File and format settings for the target.
        """
        self.config = diff_config
        self.source_config = source_config
        self.target_config = target_config

    def _normalize_headers(self, df: pl.LazyFrame) -> pl.LazyFrame:
        """Standardizes column names based on the master configuration.

        Args:
            df (pl.LazyFrame): The raw lazy dataframe.

        Returns:
            pl.LazyFrame: A dataframe with lowercased/stripped headers if enabled.
        """
        if not self.config.normalize_column_names:
            return df

        cols = df.collect_schema().names()
        rename_map = {col: col.strip().lower() for col in cols}
        return df.rename(rename_map)

    def _align_columns(self, df: pl.LazyFrame, is_source: bool = True) -> pl.LazyFrame:
        """Applies configured renames and drops ignored columns.

        Args:
            df (pl.LazyFrame): The lazy dataframe to process.
            is_source (bool): True if processing the source data, False for target.

        Returns:
            pl.LazyFrame: The structurally aligned lazy dataframe.
        """
        rename_map: dict[str, str] = {}
        to_drop: set[str] = set()
        cols = df.collect_schema().names()

        for rule in self.config.rules:
            matched_cols = [
                col
                for col in cols
                if col in rule.column_names or (rule.pattern and re.match(rule.pattern, col))
            ]

            if rule.ignore:
                to_drop.update(matched_cols)
                continue

            if (
                is_source
                and rule.rename_to
                and len(rule.column_names) == 1
                and rule.column_names[0] in cols
            ):
                rename_map[rule.column_names[0]] = rule.rename_to

        return df.drop(list(to_drop)).rename(rename_map)

    def get_dataframes(self) -> tuple[pl.LazyFrame, pl.LazyFrame]:
        """Loads and aligns both source and target datasets.

        Returns:
            tuple[pl.LazyFrame, pl.LazyFrame]: The prepared (source_df, target_df).
        """
        source_loader = LoaderFactory.get_loader(self.source_config.format)
        target_loader = LoaderFactory.get_loader(self.target_config.format)

        source_df = (
            source_loader.load(self.source_config)
            .pipe(self._normalize_headers)
            .pipe(self._align_columns, is_source=True)
        )

        target_df = (
            target_loader.load(self.target_config)
            .pipe(self._normalize_headers)
            .pipe(self._align_columns, is_source=False)
        )

        return source_df, target_df

`init(diff_config, source_config, target_config)`

Initializes the ingestor.

Parameters:

Name	Type	Description	Default
`diff_config`	`DiffConfig`	The master comparison configuration.	required
`source_config`	`SourceConfig`	File and format settings for the source.	required
`target_config`	`SourceConfig`	File and format settings for the target.	required

Source code in src/veridelta/engine.py

def __init__(
    self, diff_config: DiffConfig, source_config: SourceConfig, target_config: SourceConfig
) -> None:
    """Initializes the ingestor.

    Args:
        diff_config (DiffConfig): The master comparison configuration.
        source_config (SourceConfig): File and format settings for the source.
        target_config (SourceConfig): File and format settings for the target.
    """
    self.config = diff_config
    self.source_config = source_config
    self.target_config = target_config

`get_dataframes()`

Loads and aligns both source and target datasets.

Returns:

Type	Description
`tuple[LazyFrame, LazyFrame]`	tuple[pl.LazyFrame, pl.LazyFrame]: The prepared (source_df, target_df).

Source code in src/veridelta/engine.py

def get_dataframes(self) -> tuple[pl.LazyFrame, pl.LazyFrame]:
    """Loads and aligns both source and target datasets.

    Returns:
        tuple[pl.LazyFrame, pl.LazyFrame]: The prepared (source_df, target_df).
    """
    source_loader = LoaderFactory.get_loader(self.source_config.format)
    target_loader = LoaderFactory.get_loader(self.target_config.format)

    source_df = (
        source_loader.load(self.source_config)
        .pipe(self._normalize_headers)
        .pipe(self._align_columns, is_source=True)
    )

    target_df = (
        target_loader.load(self.target_config)
        .pipe(self._normalize_headers)
        .pipe(self._align_columns, is_source=False)
    )

    return source_df, target_df

`DiffEngine`

The core mathematical engine that evaluates differences between datasets.

Source code in src/veridelta/engine.py

class DiffEngine:
    """The core mathematical engine that evaluates differences between datasets."""

    def __init__(
        self, config: DiffConfig, source_df: pl.LazyFrame, target_df: pl.LazyFrame
    ) -> None:
        """Initializes the engine with datasets already aligned by the DataIngestor.

        Args:
            config (DiffConfig): The master validation rules configuration.
            source_df (pl.LazyFrame): The aligned 'Left' (Legacy) dataset.
            target_df (pl.LazyFrame): The aligned 'Right' (Modern) dataset.
        """
        self.config = config
        self.source = source_df
        self.target = target_df

    def _get_effective_rule(self, col_name: str) -> dict[str, Any]:
        """Resolves all rules (Specific > Pattern > Global) into a unified dictionary.

        Args:
            col_name (str): The name of the column to resolve rules for.

        Returns:
            dict[str, Any]: A flattened dictionary of operational parameters.
        """
        eff: dict[str, Any] = {
            "abs_tol": self.config.default_absolute_tolerance,
            "rel_tol": self.config.default_relative_tolerance,
            "treat_null": self.config.default_treat_null_as_equal,
            "whitespace": self.config.default_whitespace_mode,
            "null_values": self.config.default_null_values,
            "case_insensitive": False,
            "regex_replace": None,
            "value_map": None,
            "cast_to": None,
            "ignore": False,
        }

        matched_rule = None
        for rule in self.config.rules:
            if col_name in rule.column_names:
                matched_rule = rule
                break

        if not matched_rule:
            for rule in self.config.rules:
                if rule.pattern and re.match(rule.pattern, col_name):
                    matched_rule = rule
                    break

        if matched_rule:
            if matched_rule.absolute_tolerance is not None:
                eff["abs_tol"] = matched_rule.absolute_tolerance
            if matched_rule.relative_tolerance is not None:
                eff["rel_tol"] = matched_rule.relative_tolerance
            if matched_rule.treat_null_as_equal is not None:
                eff["treat_null"] = matched_rule.treat_null_as_equal
            if matched_rule.whitespace_mode is not None:
                eff["whitespace"] = matched_rule.whitespace_mode
            if matched_rule.null_values is not None:
                eff["null_values"] = matched_rule.null_values
            if matched_rule.case_insensitive is not None:
                eff["case_insensitive"] = matched_rule.case_insensitive

            eff["regex_replace"] = matched_rule.regex_replace
            eff["value_map"] = matched_rule.value_map
            eff["cast_to"] = matched_rule.cast_to
            eff["ignore"] = matched_rule.ignore

        return eff

    def _check_uniqueness(self) -> None:
        """Verifies that primary keys are unique in both datasets.

        Raises:
            DataIntegrityError: If duplicates are found in the primary keys of either
                dataset, preventing join explosions.
        """
        from veridelta.exceptions import DataIntegrityError

        pks = self.config.primary_keys

        src_pks = self.source.select(pks).collect()
        if src_pks.is_duplicated().any():
            dupes = src_pks.filter(src_pks.is_duplicated()).height
            raise DataIntegrityError(
                f"Primary keys {pks} are not unique in SOURCE dataset. "
                f"Found {dupes} duplicate rows. Clean your data before diffing."
            )

        tgt_pks = self.target.select(pks).collect()
        if tgt_pks.is_duplicated().any():
            dupes = tgt_pks.filter(tgt_pks.is_duplicated()).height
            raise DataIntegrityError(
                f"Primary keys {pks} are not unique in TARGET dataset. "
                f"Found {dupes} duplicate rows. Clean your data before diffing."
            )

    def _apply_string_rules(self, series: pl.Expr, rule: dict[str, Any]) -> pl.Expr:
        """Applies whitespace, casing, and regex cleaning to a string expression.

        Args:
            series (pl.Expr): The Polars expression representing the string column.
            rule (dict[str, Any]): The operational parameters for string transformation.

        Returns:
            pl.Expr: The transformed string expression ready for comparison.
        """
        if rule["regex_replace"]:
            for pattern, replacement in rule["regex_replace"].items():
                series = series.str.replace_all(pattern, replacement)

        mode = rule["whitespace"]
        if mode == "left":
            series = series.str.strip_chars_start()
        elif mode == "right":
            series = series.str.strip_chars_end()
        elif mode == "both":
            series = series.str.strip_chars()

        if rule["case_insensitive"]:
            series = series.str.to_lowercase()

        return series

    def _build_match_expr(self, col_name: str, rule: dict[str, Any], dtype: pl.DataType) -> pl.Expr:
        """Builds a robust comparison expression based on data type and user rules.

        Implicit Type Alignment:
            Polars is strictly typed. Comparing a Float64 to an Int64 or String raises
            a ComputeError. If a schema drift is detected between Source and Target:
            - If `strict_types=True`: The mismatch is immediately evaluated as `False`.
            - If `strict_types=False` (Default): The target column is dynamically soft-cast
              to the source's data type purely for the mathematical evaluation.

        Args:
            col_name (str): The column being compared.
            rule (dict[str, Any]): The unified rules to apply.
            dtype (pl.DataType): The data type of the source column.

        Returns:
            pl.Expr: A boolean expression evaluating to True where the row values match.
        """
        src = pl.col(f"{col_name}_source")
        tgt = pl.col(f"{col_name}_target")

        tgt_dtype = self.target.collect_schema().get(col_name)

        if dtype != tgt_dtype:
            if self.config.strict_types:
                val_match = pl.lit(False)
                if rule["treat_null"]:
                    null_match = src.is_null() & tgt.is_null()
                    return (val_match | null_match).fill_null(False)
                return val_match
            else:
                tgt = tgt.cast(dtype, strict=False)

        if rule["value_map"]:
            src = src.replace(rule["value_map"])

        if isinstance(dtype, (pl.String, pl.Utf8)):
            src = self._apply_string_rules(src, rule)
            tgt = self._apply_string_rules(tgt, rule)
            val_match = src == tgt

        elif dtype.is_numeric():
            if rule["abs_tol"] == 0.0 and rule["rel_tol"] == 0.0:
                val_match = src == tgt
            else:
                abs_diff = (tgt - src).abs()
                threshold = rule["abs_tol"] + (rule["rel_tol"] * src.abs())
                val_match = abs_diff <= threshold

        else:
            val_match = src == tgt

        if rule["treat_null"]:
            null_match = src.is_null() & tgt.is_null()
            return (val_match | null_match).fill_null(False)

        return val_match.fill_null(False)

    def _align_structure(self) -> None:
        """Perform structural normalization to reconcile asymmetrical schemas.

        Maps Source headers to Target counterparts and drops excluded fields based
        on declarative rules. This establishes the Target system's schema as the
        authoritative state, ensuring subsequent validation and comparison operate
        against a single source of truth.

        Side Effects:
            Mutates `self.source` and `self.target` to reflect the aligned structure.

        Note:
            Mandatory prerequisite for `_validate_schema`. Validating raw data
            metadata before alignment results in `ConfigError` during migrations.
        """
        src_rename: dict[str, str] = {}
        src_drop: set[str] = set()
        tgt_drop: set[str] = set()

        src_cols = self.source.collect_schema().names()
        tgt_cols = self.target.collect_schema().names()

        for rule in self.config.rules:
            matched_src = [
                col
                for col in src_cols
                if col in rule.column_names or (rule.pattern and re.match(rule.pattern, col))
            ]

            target_lookup = rule.rename_to if rule.rename_to else rule.column_names
            matched_tgt = [
                col
                for col in tgt_cols
                if col in (target_lookup if isinstance(target_lookup, list) else [target_lookup])
            ]

            if rule.ignore:
                src_drop.update(matched_src)
                tgt_drop.update(matched_tgt)
                continue

            if rule.rename_to and len(rule.column_names) == 1:
                col_name = rule.column_names[0]
                if col_name in src_cols:
                    src_rename[col_name] = rule.rename_to

        self.source = self.source.drop(list(src_drop)).rename(src_rename)
        self.target = self.target.drop(list(tgt_drop))

    def _validate_schema(self) -> None:
        """Enforces the configured SchemaMode before comparison.

        Raises:
            ConfigError: If primary keys are missing or schema constraints are violated.
        """
        source_cols = set(self.source.collect_schema().names())
        target_cols = set(self.target.collect_schema().names())
        pks = set(self.config.primary_keys)

        if not pks.issubset(source_cols):
            raise ConfigError(
                f"Primary keys missing in SOURCE after alignment: {pks - source_cols}"
            )
        if not pks.issubset(target_cols):
            raise ConfigError(f"Primary keys missing in TARGET: {pks - target_cols}")

        if self.config.schema_mode == "exact" and source_cols != target_cols:
            raise ConfigError(
                f"EXACT schema match failed.\nSource: {source_cols}\nTarget: {target_cols}"
            )

        elif self.config.schema_mode == "allow_additions":
            missing_in_target = source_cols - target_cols
            if missing_in_target:
                raise ConfigError(f"Target is missing required source columns: {missing_in_target}")

        elif self.config.schema_mode == "allow_removals":
            extra_in_target = target_cols - source_cols
            if extra_in_target:
                raise ConfigError(
                    f"Target contains unauthorized additional columns: {extra_in_target}"
                )

    def run(self) -> DiffSummary:
        """Execute the end-to-end dataset comparison pipeline lazily.

        Builds an optimized Polars computation graph (DAG) to guarantee deterministic
        alignment, preventing compute errors and memory exhaustion on large datasets.
        Data is only materialized into memory when absolutely necessary for execution.

        Execution Pipeline:
            1. Structural Alignment: Maps and prunes schemas to establish the
               Target as the authoritative structural contract.
            2. Validation & Integrity: Asserts primary key existence and uniqueness
               (triggering a localized collection), and enforces the `SchemaMode`.
            3. Lazy Graph Construction: Builds the computation DAG for semantic
               normalization (regex sanitization) and type coercion.
            4. Relational Joins: Formulates the lazy anti-joins ('Added', 'Removed')
               and inner-joins ('Changed') to isolate discrepancies.
            5. Graph Execution: Executes the computation DAG via `.collect()` to
               evaluate vectorized match expressions and compute exact row counts.
            6. Artifact Persistence: Exports the materialized discrepancy dataframes
               to the configured storage backend, if requested.

        Returns:
            DiffSummary: Execution report detailing match status, discrepancy counts,
                and column-level drift metrics.

        Raises:
            ConfigError: If schema constraints or primary keys are violated post-alignment.
            DataIntegrityError: If duplicate primary keys prevent deterministic joins.
            NotImplementedError: If the requested artifact export format is unsupported.
        """
        self._align_structure()
        self._validate_schema()
        self._check_uniqueness()

        src_cols = self.source.collect_schema().names()
        tgt_cols = self.target.collect_schema().names()

        for col in src_cols:
            rule = self._get_effective_rule(col)

            if rule["null_values"]:
                null_list = rule["null_values"]
                if col in src_cols:
                    self.source = self.source.with_columns(
                        pl.when(pl.col(col).is_in(null_list))
                        .then(None)
                        .otherwise(pl.col(col))
                        .alias(col)
                    )
                if col in tgt_cols:
                    self.target = self.target.with_columns(
                        pl.when(pl.col(col).is_in(null_list))
                        .then(None)
                        .otherwise(pl.col(col))
                        .alias(col)
                    )

            if rule["regex_replace"]:
                for pattern, replacement in rule["regex_replace"].items():
                    if col in src_cols and isinstance(
                        self.source.collect_schema().get(col), (pl.String, pl.Utf8)
                    ):
                        self.source = self.source.with_columns(
                            pl.col(col).str.replace_all(pattern, replacement)
                        )
                    if col in tgt_cols and isinstance(
                        self.target.collect_schema().get(col), (pl.String, pl.Utf8)
                    ):
                        self.target = self.target.with_columns(
                            pl.col(col).str.replace_all(pattern, replacement)
                        )

            if rule["cast_to"]:
                dtype = getattr(pl, rule["cast_to"], None)
                if dtype:
                    if col in src_cols:
                        self.source = self.source.with_columns(pl.col(col).cast(dtype))
                    if col in tgt_cols:
                        self.target = self.target.with_columns(pl.col(col).cast(dtype))

        added_lazy = self.target.join(self.source, on=self.config.primary_keys, how="anti")
        removed_lazy = self.source.join(self.target, on=self.config.primary_keys, how="anti")

        # Re-fetch schema names in case rules altered them
        src_cols_final = self.source.collect_schema().names()
        tgt_cols_final = self.target.collect_schema().names()

        src_renamed = self.source.rename(
            {col: f"{col}_source" for col in src_cols_final if col not in self.config.primary_keys}
        )
        tgt_renamed = self.target.rename(
            {col: f"{col}_target" for col in tgt_cols_final if col not in self.config.primary_keys}
        )
        common_lazy = src_renamed.join(tgt_renamed, on=self.config.primary_keys, how="inner")

        match_expressions: list[pl.Expr] = []
        match_cols: list[str] = []

        for col in src_cols_final:
            if col in self.config.primary_keys or col not in tgt_cols_final:
                continue

            rule = self._get_effective_rule(col)
            if rule["ignore"]:
                continue

            dtype = self.source.collect_schema()[col]
            expr = self._build_match_expr(col, rule, dtype).alias(f"{col}_is_match")
            match_expressions.append(expr)
            match_cols.append(f"{col}_is_match")

        added_df = added_lazy.collect()
        removed_df = removed_lazy.collect()

        changed_count = 0
        changed_df = pl.DataFrame()
        column_mismatches: dict[str, int] = {}

        if match_expressions:
            evaluated_lazy = common_lazy.with_columns(match_expressions)
            all_matched = pl.all_horizontal(match_cols)
            changed_lazy = evaluated_lazy.filter(~all_matched)

            changed_df = changed_lazy.collect()
            changed_count = changed_df.height

            if changed_count > 0:
                mismatch_exprs = [
                    (~pl.col(c)).sum().alias(c.replace("_is_match", "")) for c in match_cols
                ]
                raw_counts = changed_df.select(mismatch_exprs).to_dicts()[0]
                column_mismatches = {k: v for k, v in raw_counts.items() if v > 0}

        pk_col = self.config.primary_keys[0]
        src_total = self.source.select(pl.col(pk_col).count()).collect().item()
        tgt_total = self.target.select(pl.col(pk_col).count()).collect().item()

        mismatch_ratio = (added_df.height + removed_df.height + changed_count) / max(src_total, 1)
        is_match = mismatch_ratio <= self.config.threshold

        output_path_str = getattr(self.config, "output_path", None)
        if isinstance(output_path_str, str):
            out_dir = Path(output_path_str)
            out_dir.mkdir(parents=True, exist_ok=True)
            fmt = getattr(self.config, "output_format", "parquet")

            def _export_artifact(df: pl.DataFrame, name: str) -> None:
                if df.height == 0:
                    return
                file_path = out_dir / f"{name}.{fmt}"
                if fmt == "csv":
                    df.write_csv(file_path)
                elif fmt == "parquet":
                    df.write_parquet(file_path)
                else:
                    raise NotImplementedError(
                        f"Export support for format '{fmt}' is not yet implemented."
                    )

            _export_artifact(added_df, "added_rows")
            _export_artifact(removed_df, "removed_rows")
            _export_artifact(changed_df, "changed_rows")

        return DiffSummary(
            total_rows_source=src_total,
            total_rows_target=tgt_total,
            added_count=added_df.height,
            removed_count=removed_df.height,
            changed_count=changed_count,
            column_mismatches=column_mismatches,
            is_match=is_match,
            report_limit=self.config.report_top_columns_limit,
        )

`init(config, source_df, target_df)`

Initializes the engine with datasets already aligned by the DataIngestor.

Parameters:

Name	Type	Description	Default
`config`	`DiffConfig`	The master validation rules configuration.	required
`source_df`	`LazyFrame`	The aligned 'Left' (Legacy) dataset.	required
`target_df`	`LazyFrame`	The aligned 'Right' (Modern) dataset.	required

Source code in src/veridelta/engine.py

def __init__(
    self, config: DiffConfig, source_df: pl.LazyFrame, target_df: pl.LazyFrame
) -> None:
    """Initializes the engine with datasets already aligned by the DataIngestor.

    Args:
        config (DiffConfig): The master validation rules configuration.
        source_df (pl.LazyFrame): The aligned 'Left' (Legacy) dataset.
        target_df (pl.LazyFrame): The aligned 'Right' (Modern) dataset.
    """
    self.config = config
    self.source = source_df
    self.target = target_df

`run()`

Execute the end-to-end dataset comparison pipeline lazily.

Builds an optimized Polars computation graph (DAG) to guarantee deterministic alignment, preventing compute errors and memory exhaustion on large datasets. Data is only materialized into memory when absolutely necessary for execution.

Execution Pipeline

Structural Alignment: Maps and prunes schemas to establish the Target as the authoritative structural contract.
Validation & Integrity: Asserts primary key existence and uniqueness (triggering a localized collection), and enforces the SchemaMode.
Lazy Graph Construction: Builds the computation DAG for semantic normalization (regex sanitization) and type coercion.
Relational Joins: Formulates the lazy anti-joins ('Added', 'Removed') and inner-joins ('Changed') to isolate discrepancies.
Graph Execution: Executes the computation DAG via .collect() to evaluate vectorized match expressions and compute exact row counts.
Artifact Persistence: Exports the materialized discrepancy dataframes to the configured storage backend, if requested.

Returns:

Name	Type	Description
`DiffSummary`	`DiffSummary`	Execution report detailing match status, discrepancy counts, and column-level drift metrics.

Raises:

Type	Description
`ConfigError`	If schema constraints or primary keys are violated post-alignment.
`DataIntegrityError`	If duplicate primary keys prevent deterministic joins.
`NotImplementedError`	If the requested artifact export format is unsupported.

Source code in src/veridelta/engine.py

def run(self) -> DiffSummary:
    """Execute the end-to-end dataset comparison pipeline lazily.

    Builds an optimized Polars computation graph (DAG) to guarantee deterministic
    alignment, preventing compute errors and memory exhaustion on large datasets.
    Data is only materialized into memory when absolutely necessary for execution.

    Execution Pipeline:
        1. Structural Alignment: Maps and prunes schemas to establish the
           Target as the authoritative structural contract.
        2. Validation & Integrity: Asserts primary key existence and uniqueness
           (triggering a localized collection), and enforces the `SchemaMode`.
        3. Lazy Graph Construction: Builds the computation DAG for semantic
           normalization (regex sanitization) and type coercion.
        4. Relational Joins: Formulates the lazy anti-joins ('Added', 'Removed')
           and inner-joins ('Changed') to isolate discrepancies.
        5. Graph Execution: Executes the computation DAG via `.collect()` to
           evaluate vectorized match expressions and compute exact row counts.
        6. Artifact Persistence: Exports the materialized discrepancy dataframes
           to the configured storage backend, if requested.

    Returns:
        DiffSummary: Execution report detailing match status, discrepancy counts,
            and column-level drift metrics.

    Raises:
        ConfigError: If schema constraints or primary keys are violated post-alignment.
        DataIntegrityError: If duplicate primary keys prevent deterministic joins.
        NotImplementedError: If the requested artifact export format is unsupported.
    """
    self._align_structure()
    self._validate_schema()
    self._check_uniqueness()

    src_cols = self.source.collect_schema().names()
    tgt_cols = self.target.collect_schema().names()

    for col in src_cols:
        rule = self._get_effective_rule(col)

        if rule["null_values"]:
            null_list = rule["null_values"]
            if col in src_cols:
                self.source = self.source.with_columns(
                    pl.when(pl.col(col).is_in(null_list))
                    .then(None)
                    .otherwise(pl.col(col))
                    .alias(col)
                )
            if col in tgt_cols:
                self.target = self.target.with_columns(
                    pl.when(pl.col(col).is_in(null_list))
                    .then(None)
                    .otherwise(pl.col(col))
                    .alias(col)
                )

        if rule["regex_replace"]:
            for pattern, replacement in rule["regex_replace"].items():
                if col in src_cols and isinstance(
                    self.source.collect_schema().get(col), (pl.String, pl.Utf8)
                ):
                    self.source = self.source.with_columns(
                        pl.col(col).str.replace_all(pattern, replacement)
                    )
                if col in tgt_cols and isinstance(
                    self.target.collect_schema().get(col), (pl.String, pl.Utf8)
                ):
                    self.target = self.target.with_columns(
                        pl.col(col).str.replace_all(pattern, replacement)
                    )

        if rule["cast_to"]:
            dtype = getattr(pl, rule["cast_to"], None)
            if dtype:
                if col in src_cols:
                    self.source = self.source.with_columns(pl.col(col).cast(dtype))
                if col in tgt_cols:
                    self.target = self.target.with_columns(pl.col(col).cast(dtype))

    added_lazy = self.target.join(self.source, on=self.config.primary_keys, how="anti")
    removed_lazy = self.source.join(self.target, on=self.config.primary_keys, how="anti")

    # Re-fetch schema names in case rules altered them
    src_cols_final = self.source.collect_schema().names()
    tgt_cols_final = self.target.collect_schema().names()

    src_renamed = self.source.rename(
        {col: f"{col}_source" for col in src_cols_final if col not in self.config.primary_keys}
    )
    tgt_renamed = self.target.rename(
        {col: f"{col}_target" for col in tgt_cols_final if col not in self.config.primary_keys}
    )
    common_lazy = src_renamed.join(tgt_renamed, on=self.config.primary_keys, how="inner")

    match_expressions: list[pl.Expr] = []
    match_cols: list[str] = []

    for col in src_cols_final:
        if col in self.config.primary_keys or col not in tgt_cols_final:
            continue

        rule = self._get_effective_rule(col)
        if rule["ignore"]:
            continue

        dtype = self.source.collect_schema()[col]
        expr = self._build_match_expr(col, rule, dtype).alias(f"{col}_is_match")
        match_expressions.append(expr)
        match_cols.append(f"{col}_is_match")

    added_df = added_lazy.collect()
    removed_df = removed_lazy.collect()

    changed_count = 0
    changed_df = pl.DataFrame()
    column_mismatches: dict[str, int] = {}

    if match_expressions:
        evaluated_lazy = common_lazy.with_columns(match_expressions)
        all_matched = pl.all_horizontal(match_cols)
        changed_lazy = evaluated_lazy.filter(~all_matched)

        changed_df = changed_lazy.collect()
        changed_count = changed_df.height

        if changed_count > 0:
            mismatch_exprs = [
                (~pl.col(c)).sum().alias(c.replace("_is_match", "")) for c in match_cols
            ]
            raw_counts = changed_df.select(mismatch_exprs).to_dicts()[0]
            column_mismatches = {k: v for k, v in raw_counts.items() if v > 0}

    pk_col = self.config.primary_keys[0]
    src_total = self.source.select(pl.col(pk_col).count()).collect().item()
    tgt_total = self.target.select(pl.col(pk_col).count()).collect().item()

    mismatch_ratio = (added_df.height + removed_df.height + changed_count) / max(src_total, 1)
    is_match = mismatch_ratio <= self.config.threshold

    output_path_str = getattr(self.config, "output_path", None)
    if isinstance(output_path_str, str):
        out_dir = Path(output_path_str)
        out_dir.mkdir(parents=True, exist_ok=True)
        fmt = getattr(self.config, "output_format", "parquet")

        def _export_artifact(df: pl.DataFrame, name: str) -> None:
            if df.height == 0:
                return
            file_path = out_dir / f"{name}.{fmt}"
            if fmt == "csv":
                df.write_csv(file_path)
            elif fmt == "parquet":
                df.write_parquet(file_path)
            else:
                raise NotImplementedError(
                    f"Export support for format '{fmt}' is not yet implemented."
                )

        _export_artifact(added_df, "added_rows")
        _export_artifact(removed_df, "removed_rows")
        _export_artifact(changed_df, "changed_rows")

    return DiffSummary(
        total_rows_source=src_total,
        total_rows_target=tgt_total,
        added_count=added_df.height,
        removed_count=removed_df.height,
        changed_count=changed_count,
        column_mismatches=column_mismatches,
        is_match=is_match,
        report_limit=self.config.report_top_columns_limit,
    )

`LoaderFactory`

Factory to return the appropriate loader based on the configured SourceType.

Source code in src/veridelta/engine.py

class LoaderFactory:
    """Factory to return the appropriate loader based on the configured SourceType."""

    _loaders: ClassVar[dict[str, BaseLoader]] = {
        "csv": CSVLoader(),
        "parquet": ParquetLoader(),
    }

    @classmethod
    def get_loader(cls, source_type: str) -> BaseLoader:
        """Retrieves the correct loader instance for the given data format.

        Args:
            source_type (str): The format identifier (e.g., 'csv', 'parquet').

        Returns:
            BaseLoader: An instantiated data loader.

        Raises:
            NotImplementedError: If the requested format is not yet supported.
        """
        loader = cls._loaders.get(source_type)
        if not loader:
            raise NotImplementedError(
                f"Support for '{source_type}' is planned but not yet implemented."
            )
        return loader

`get_loader(source_type)` `classmethod`

Retrieves the correct loader instance for the given data format.

Parameters:

Name	Type	Description	Default
`source_type`	`str`	The format identifier (e.g., 'csv', 'parquet').	required

Returns:

Name	Type	Description
`BaseLoader`	`BaseLoader`	An instantiated data loader.

Raises:

Type	Description
`NotImplementedError`	If the requested format is not yet supported.

Source code in src/veridelta/engine.py

@classmethod
def get_loader(cls, source_type: str) -> BaseLoader:
    """Retrieves the correct loader instance for the given data format.

    Args:
        source_type (str): The format identifier (e.g., 'csv', 'parquet').

    Returns:
        BaseLoader: An instantiated data loader.

    Raises:
        NotImplementedError: If the requested format is not yet supported.
    """
    loader = cls._loaders.get(source_type)
    if not loader:
        raise NotImplementedError(
            f"Support for '{source_type}' is planned but not yet implemented."
        )
    return loader

`ParquetLoader`

Bases: BaseLoader

Loader for Parquet files utilizing the Polars Parquet engine.

Source code in src/veridelta/engine.py

class ParquetLoader(BaseLoader):
    """Loader for Parquet files utilizing the Polars Parquet engine."""

    def load(self, config: SourceConfig) -> pl.LazyFrame:
        """Loads a Parquet file into a Polars LazyFrame.

        Args:
            config (SourceConfig): The source configuration. Extra options are
                passed directly to `pl.scan_parquet`.

        Returns:
            pl.LazyFrame: The lazy dataset graph.
        """
        return pl.scan_parquet(config.path, **config.options)

`load(config)`

Loads a Parquet file into a Polars LazyFrame.

Parameters:

Name	Type	Description	Default
`config`	`SourceConfig`	The source configuration. Extra options are passed directly to `pl.scan_parquet`.	required

Returns:

Type	Description
`LazyFrame`	pl.LazyFrame: The lazy dataset graph.

Source code in src/veridelta/engine.py

def load(self, config: SourceConfig) -> pl.LazyFrame:
    """Loads a Parquet file into a Polars LazyFrame.

    Args:
        config (SourceConfig): The source configuration. Extra options are
            passed directly to `pl.scan_parquet`.

    Returns:
        pl.LazyFrame: The lazy dataset graph.
    """
    return pl.scan_parquet(config.path, **config.options)

Configuration Parser

Utilities for loading, parsing, and validating YAML files into strictly typed configuration objects.

Configuration parsing and validation from YAML files.

This module acts as the bridge between user-defined YAML configurations and the strict Pydantic models required by the execution engine.

`load_config(path)`

Loads and validates a Veridelta configuration from a YAML file.

The parser extracts the explicit source and target definition blocks, then evaluates all remaining root-level YAML parameters as the master DiffConfig.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	The file system path to the YAML configuration.	required

Returns:

Type	Description
`tuple[DiffConfig, SourceConfig, SourceConfig]`	tuple[DiffConfig, SourceConfig, SourceConfig]: A tuple containing the validated master configuration, source configuration, and target configuration objects respectively.

Raises:

Type	Description
`ConfigError`	If the file cannot be located, contains invalid YAML syntax, lacks the mandatory source/target blocks, or violates the strict Pydantic schema definitions.

Source code in src/veridelta/config.py

def load_config(path: str | Path) -> tuple[DiffConfig, SourceConfig, SourceConfig]:
    """Loads and validates a Veridelta configuration from a YAML file.

    The parser extracts the explicit `source` and `target` definition blocks,
    then evaluates all remaining root-level YAML parameters as the master
    `DiffConfig`.

    Args:
        path (str | Path): The file system path to the YAML configuration.

    Returns:
        tuple[DiffConfig, SourceConfig, SourceConfig]: A tuple containing the
            validated master configuration, source configuration, and target
            configuration objects respectively.

    Raises:
        ConfigError: If the file cannot be located, contains invalid YAML syntax,
            lacks the mandatory source/target blocks, or violates the strict
            Pydantic schema definitions.
    """
    file_path = Path(path)

    if not file_path.is_file():
        raise ConfigError(f"Configuration file not found or is not a file: {file_path.absolute()}")

    try:
        with file_path.open("r", encoding="utf-8") as f:
            parsed_yaml: Any = yaml.safe_load(f)
    except yaml.YAMLError as yaml_err:
        raise ConfigError(f"Failed to parse YAML file:\n{yaml_err}") from yaml_err

    if not isinstance(parsed_yaml, dict):
        raise ConfigError("Invalid YAML structure: Root element must be a dictionary.")

    raw_config = cast("dict[str, Any]", parsed_yaml)
    if "source" not in raw_config or "target" not in raw_config:
        raise ConfigError("Configuration must contain both 'source' and 'target' blocks.")

    try:
        raw_source = raw_config.pop("source")
        raw_target = raw_config.pop("target")

        source_cfg = SourceConfig.model_validate(raw_source)
        target_cfg = SourceConfig.model_validate(raw_target)
        diff_cfg = DiffConfig.model_validate(raw_config)

        return diff_cfg, source_cfg, target_cfg

    except ValidationError as e:
        error_msg = "Configuration Validation Failed:\n"
        for validation_error in e.errors():
            location = " -> ".join(str(loc) for loc in validation_error["loc"])
            error_msg += f"  - [{location}]: {validation_error['msg']}\n"
        raise ConfigError(error_msg) from e

Exceptions

The custom exception hierarchy. Consumers of the Python API should handle these specific errors to manage pipeline failures gracefully without silencing native Python runtime panics.

Custom exceptions for Veridelta operations.

This module defines the core exception hierarchy used throughout the Veridelta framework. Consumers of the Python API can catch the base VerideltaError to safely handle all framework-specific failures.

`ConfigError`

Bases: VerideltaError

Raised when configuration validation or schema enforcement fails.

This is triggered during pipeline initialization or schema validation if mandatory parameters (like primary keys) are missing, or if strict schema constraints (e.g., allow_removals, exact) are violated by the provided datasets.

Source code in src/veridelta/exceptions.py

class ConfigError(VerideltaError):
    """Raised when configuration validation or schema enforcement fails.

    This is triggered during pipeline initialization or schema validation
    if mandatory parameters (like primary keys) are missing, or if strict
    schema constraints (e.g., `allow_removals`, `exact`) are violated by
    the provided datasets.
    """

`DataIntegrityError`

Bases: VerideltaError

Raised when foundational data assumptions are violated.

This is typically raised during the pre-evaluation phase if primary keys are not unique within either dataset. Halting execution on this error prevents catastrophic join explosions and Out-Of-Memory (OOM) crashes during the Polars evaluation phase.

Source code in src/veridelta/exceptions.py

class DataIntegrityError(VerideltaError):
    """Raised when foundational data assumptions are violated.

    This is typically raised during the pre-evaluation phase if primary
    keys are not unique within either dataset. Halting execution on this
    error prevents catastrophic join explosions and Out-Of-Memory (OOM)
    crashes during the Polars evaluation phase.
    """

`VerideltaError`

Bases: Exception

Base exception for all Veridelta-specific errors.

Consumers should catch this exception to handle pipeline validation failures gracefully without silencing standard Python runtime errors (like MemoryError or ValueError).

Source code in src/veridelta/exceptions.py

class VerideltaError(Exception):
    """Base exception for all Veridelta-specific errors.

    Consumers should catch this exception to handle pipeline validation
    failures gracefully without silencing standard Python runtime errors
    (like `MemoryError` or `ValueError`).
    """

Datasets

Built-in data utilities with network-resilient caching for testing, onboarding, and tutorials.

Built-in datasets for Veridelta testing and quickstart examples.

This module provides utilities to securely download, cache, and load sample datasets used in Veridelta's documentation and tutorials.

`load_nyc_taxi()`

Loads the NYC Taxi sample dataset.

Downloads the dataset from the official Veridelta repository and caches it locally. If the cached file is corrupted, it automatically evicts it and attempts a fresh download. Enforces a 15-second timeout.

Returns:

Type	Description
`DataFrame`	pl.DataFrame: A Polars DataFrame containing the NYC Taxi sample data.

Raises:

Type	Description
`RuntimeError`	If the download fails due to network or routing issues.

Source code in src/veridelta/datasets.py

def load_nyc_taxi() -> pl.DataFrame:
    """Loads the NYC Taxi sample dataset.

    Downloads the dataset from the official Veridelta repository and caches it
    locally. If the cached file is corrupted, it automatically evicts it and
    attempts a fresh download. Enforces a 15-second timeout.

    Returns:
        pl.DataFrame: A Polars DataFrame containing the NYC Taxi sample data.

    Raises:
        RuntimeError: If the download fails due to network or routing issues.
    """
    cache_path = _get_cache_dir() / "sample_taxi_data.parquet"

    def _download_file() -> None:
        logger.warning(f"Downloading NYC Taxi dataset to {cache_path}...")
        try:
            req = urllib.request.Request(_TAXI_URL)
            with (
                urllib.request.urlopen(req, timeout=15.0) as response,
                open(cache_path, "wb") as out_file,
            ):
                shutil.copyfileobj(response, out_file)

        except urllib.error.URLError as e:
            if cache_path.exists():
                cache_path.unlink()
            raise RuntimeError(
                f"Failed to download Veridelta sample dataset. "
                f"Check your internet connection or the URL. Error: {e}"
            ) from e

    if not cache_path.exists():
        _download_file()

    try:
        return pl.read_parquet(cache_path)
    except pl.exceptions.PolarsError:
        logger.warning("Cached dataset is corrupted. Evicting and re-downloading...")
        if cache_path.exists():
            cache_path.unlink()

        _download_file()
        return pl.read_parquet(cache_path)

API Reference

Configuration Models

SchemaMode = Literal['exact', 'allow_additions', 'allow_removals', 'intersection'] module-attribute

SourceType = Literal['csv', 'json', 'parquet', 'fixed_width', 'netcdf', 'shapefile', 'geopackage', 'excel', 'sql', 'delta', 'avro', 'xml', 'arrow'] module-attribute

WhitespaceMode = Literal['none', 'left', 'right', 'both'] module-attribute

DiffConfig

apply_schema_normalization()

DiffRule

validate_pattern(v) classmethod

validate_regex_replace(v) classmethod

DiffSummary

is_perfect_match property

match_rate_percentage property

mismatch_ratio property

report_summary property

total_mismatches property

volume_shift property

SourceConfig

The Engine

BaseLoader

load(config) abstractmethod

CSVLoader

load(config)

DataIngestor

__init__(diff_config, source_config, target_config)

get_dataframes()

DiffEngine

__init__(config, source_df, target_df)

run()

LoaderFactory

get_loader(source_type) classmethod

ParquetLoader

load(config)

Configuration Parser

load_config(path)

Exceptions

ConfigError

DataIntegrityError

VerideltaError

Datasets

load_nyc_taxi()

`SchemaMode = Literal['exact', 'allow_additions', 'allow_removals', 'intersection']` `module-attribute`

`SourceType = Literal['csv', 'json', 'parquet', 'fixed_width', 'netcdf', 'shapefile', 'geopackage', 'excel', 'sql', 'delta', 'avro', 'xml', 'arrow']` `module-attribute`

`WhitespaceMode = Literal['none', 'left', 'right', 'both']` `module-attribute`

`DiffConfig`

`apply_schema_normalization()`

`DiffRule`

`validate_pattern(v)` `classmethod`

`validate_regex_replace(v)` `classmethod`

`DiffSummary`

`is_perfect_match` `property`

`match_rate_percentage` `property`

`mismatch_ratio` `property`

`report_summary` `property`

`total_mismatches` `property`

`volume_shift` `property`

`SourceConfig`

`BaseLoader`

`load(config)` `abstractmethod`

`CSVLoader`

`load(config)`

`DataIngestor`

`init(diff_config, source_config, target_config)`

`get_dataframes()`

`DiffEngine`

`init(config, source_df, target_df)`

`run()`

`LoaderFactory`

`get_loader(source_type)` `classmethod`

`ParquetLoader`

`load(config)`

`load_config(path)`

`ConfigError`

`DataIntegrityError`

`VerideltaError`

`load_nyc_taxi()`