API Documentation

datajudge allows to assess whether data from database complies with referenceinformation.

BetweenRequirement

BetweenRequirement(data_source: DataSource, data_source2: DataSource, date_column: str | None = None, date_column2: str | None = None)

Bases: Requirement

Source code in src/datajudge/requirements.py

def __init__(
    self,
    data_source: DataSource,
    data_source2: DataSource,
    date_column: str | None = None,
    date_column2: str | None = None,
):
    self._data_source = data_source
    self._data_source2 = data_source2
    self._ref = DataReference(self._data_source)
    self._ref2 = DataReference(self._data_source2)
    self._date_column = date_column
    self._date_column2 = date_column2
    super().__init__()

add_column_subset_constraint

add_column_subset_constraint(name: str | None = None, cache_size=None) -> None

Columns of first table are subset of second table.

Source code in src/datajudge/requirements.py

def add_column_subset_constraint(
    self, name: str | None = None, cache_size=None
) -> None:
    """Columns of first table are subset of second table."""
    self._constraints.append(
        column_constraints.ColumnSubset(
            self._ref, ref2=self._ref2, name=name, cache_size=cache_size
        )
    )

add_column_superset_constraint

add_column_superset_constraint(name: str | None = None, cache_size=None) -> None

Columns of first table are superset of columns of second table.

Source code in src/datajudge/requirements.py

def add_column_superset_constraint(
    self, name: str | None = None, cache_size=None
) -> None:
    """Columns of first table are superset of columns of second table."""
    self._constraints.append(
        column_constraints.ColumnSuperset(
            self._ref, ref2=self._ref2, name=name, cache_size=cache_size
        )
    )

add_column_type_constraint

add_column_type_constraint(column1: str, column2: str, name: str | None = None, cache_size=None) -> None

Check that the columns have the same type.

Source code in src/datajudge/requirements.py

def add_column_type_constraint(
    self,
    column1: str,
    column2: str,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Check that the columns have the same type."""
    ref1 = DataReference(self._data_source, [column1])
    ref2 = DataReference(self._data_source2, [column2])
    self._constraints.append(
        column_constraints.ColumnType(
            ref1, ref2=ref2, name=name, cache_size=cache_size
        )
    )

add_date_max_constraint

add_date_max_constraint(column1: str, column2: str, use_upper_bound_reference: bool = True, column_type: str = 'date', condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Compare date max of first table to date max of second table.

The used columns of both tables need to be of the same type.

For more information on column_type values, see add_column_type_constraint.

If use_upper_bound_reference, the max of the first table has to be smaller or equal to the max of the second table. If not use_upper_bound_reference, the max of the first table has to be greater or equal to the max of the second table.

Source code in src/datajudge/requirements.py

def add_date_max_constraint(
    self,
    column1: str,
    column2: str,
    use_upper_bound_reference: bool = True,
    column_type: str = "date",
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Compare date max of first table to date max of second table.

    The used columns of both tables need to be of the same type.

    For more information on ``column_type`` values, see ``add_column_type_constraint``.

    If ``use_upper_bound_reference``, the max of the first table has to be
    smaller or equal to the max of the second table.
    If not ``use_upper_bound_reference``, the max of the first table has to
    be greater or equal to the max of the second table.
    """
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        date_constraints.DateMax(
            ref,
            ref2=ref2,
            use_upper_bound_reference=use_upper_bound_reference,
            column_type=column_type,
            name=name,
            cache_size=cache_size,
        )
    )

add_date_min_constraint

add_date_min_constraint(column1: str, column2: str, use_lower_bound_reference: bool = True, column_type: str = 'date', condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Ensure date min of first table is greater or equal date min of second table.

The used columns of both tables need to be of the same type.

For more information on column_type values, see add_column_type_constraint.

If use_lower_bound_reference, the min of the first table has to be greater or equal to the min of the second table. If not use_upper_bound_reference, the min of the first table has to be smaller or equal to the min of the second table.

Source code in src/datajudge/requirements.py

def add_date_min_constraint(
    self,
    column1: str,
    column2: str,
    use_lower_bound_reference: bool = True,
    column_type: str = "date",
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Ensure date min of first table is greater or equal date min of second table.

    The used columns of both tables need to be of the same type.

    For more information on ``column_type`` values, see ``add_column_type_constraint``.

    If ``use_lower_bound_reference``, the min of the first table has to be
    greater or equal to the min of the second table.
    If not ``use_upper_bound_reference``, the min of the first table has to
    be smaller or equal to the min of the second table.
    """
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        date_constraints.DateMin(
            ref,
            ref2=ref2,
            use_lower_bound_reference=use_lower_bound_reference,
            column_type=column_type,
            name=name,
            cache_size=cache_size,
        )
    )

add_ks_2sample_constraint

add_ks_2sample_constraint(column1: str, column2: str, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, significance_level: float = 0.05, cache_size=None) -> None

Apply the so-called two-sample Kolmogorov-Smirnov test to the distributions of the two given columns.

The constraint is fulfilled, when the resulting p-value of the test is higher than the significance level (default is 0.05, i.e., 5%). The significance_level must be a value between 0.0 and 1.0.

Source code in src/datajudge/requirements.py

def add_ks_2sample_constraint(
    self,
    column1: str,
    column2: str,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    significance_level: float = 0.05,
    cache_size=None,
) -> None:
    """
    Apply the so-called two-sample Kolmogorov-Smirnov test to the distributions of the two given columns.

    The constraint is fulfilled, when the resulting p-value of the test is higher than the significance level
    (default is 0.05, i.e., 5%).
    The significance_level must be a value between 0.0 and 1.0.
    """
    if not column1 or not column2:
        raise ValueError(
            "Column names have to be given for this test's functionality."
        )

    if significance_level <= 0.0 or significance_level > 1.0:
        raise ValueError(
            "The requested significance level has to be in ``(0.0, 1.0]``. Default is 0.05."
        )

    ref = DataReference(self._data_source, [column1], condition=condition1)
    ref2 = DataReference(self._data_source2, [column2], condition=condition2)
    self._constraints.append(
        stats_constraints.KolmogorovSmirnov2Sample(
            ref,
            ref2,
            significance_level,
            name=name,
            cache_size=cache_size,
        )
    )

add_max_null_fraction_constraint

add_max_null_fraction_constraint(column1: str, column2: str, max_relative_deviation: float, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the fraction of NULL values of one is at most that of the other.

Given that column2's underlying data has a fraction q of NULL values, the max_relative_deviation parameter allows column1's underlying data to have a fraction (1 + max_relative_deviation) * q of NULL values.

Source code in src/datajudge/requirements.py

def add_max_null_fraction_constraint(
    self,
    column1: str,
    column2: str,
    max_relative_deviation: float,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Assert that the fraction of ``NULL`` values of one is at most that of the other.

    Given that ``column2``'s underlying data has a fraction ``q`` of ``NULL`` values, the
    ``max_relative_deviation`` parameter allows ``column1``'s underlying data to have a
    fraction ``(1 + max_relative_deviation) * q`` of ``NULL`` values.
    """  # noqa: D301
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        miscs_constraints.MaxNullFraction(
            ref,
            ref2=ref2,
            max_relative_deviation=max_relative_deviation,
            cache_size=cache_size,
        )
    )

add_n_rows_equality_constraint

add_n_rows_equality_constraint(condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_n_rows_equality_constraint(
    self,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    ref = DataReference(self._data_source, condition=condition1)
    ref2 = DataReference(self._data_source2, condition=condition2)
    self._constraints.append(
        nrows_constraints.NRowsEquality(
            ref, ref2=ref2, name=name, cache_size=cache_size
        )
    )

add_n_rows_max_gain_constraint

add_n_rows_max_gain_constraint(constant_max_relative_gain: float | None = None, date_range_gain_deviation: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the number of rows hasn't grown by more than expected.

In particular, assert that

\[n^{rows}_1 \leq n^{rows}_2 \cdot (1 + \text{cmrg})\]

See readme for more information on constant_max_relative_gain.

Source code in src/datajudge/requirements.py

def add_n_rows_max_gain_constraint(
    self,
    constant_max_relative_gain: float | None = None,
    date_range_gain_deviation: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    r"""Assert that the number of rows hasn't grown by more than expected.

    In particular, assert that

    $$n^{rows}_1 \leq n^{rows}_2 \cdot (1 + \text{cmrg})$$

    See readme for more information on ``constant_max_relative_gain``.
    """
    max_relative_gain_getter = self._get_deviation_getter(
        constant_max_relative_gain, date_range_gain_deviation
    )
    ref = DataReference(self._data_source, condition=condition1)
    ref2 = DataReference(self._data_source2, condition=condition2)
    self._constraints.append(
        nrows_constraints.NRowsMaxGain(
            ref,
            ref2,
            max_relative_gain_getter,
            name=name,
            cache_size=cache_size,
        )
    )

add_n_rows_max_loss_constraint

add_n_rows_max_loss_constraint(constant_max_relative_loss: float | None = None, date_range_loss_deviation: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the number of rows hasn't decreased too much.

In particular, assert that

\[n^{rows}_1 \geq n^{rows}_2 \cdot (1 - \text{cmrl})\]

See readme for more information on constant_max_relative_loss.

Source code in src/datajudge/requirements.py

def add_n_rows_max_loss_constraint(
    self,
    constant_max_relative_loss: float | None = None,
    date_range_loss_deviation: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    r"""Assert that the number of rows hasn't decreased too much.

    In particular, assert that

    $$n^{rows}_1 \geq n^{rows}_2 \cdot (1 - \text{cmrl})$$

    See readme for more information on ``constant_max_relative_loss``.
    """
    max_relative_loss_getter = self._get_deviation_getter(
        constant_max_relative_loss, date_range_loss_deviation
    )
    ref = DataReference(self._data_source, condition=condition1)
    ref2 = DataReference(self._data_source2, condition=condition2)
    self._constraints.append(
        nrows_constraints.NRowsMaxLoss(
            ref,
            ref2,
            max_relative_loss_getter,
            name=name,
            cache_size=cache_size,
        )
    )

add_n_rows_min_gain_constraint

add_n_rows_min_gain_constraint(constant_min_relative_gain: float | None = None, date_range_gain_deviation: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the number of rows hasn't grown less than expected.

In particular, assert that

\[n^{rows}_1 \geq n^{rows}_2 \cdot (1 + \text{cmrg})\]

See readme for more information on constant_min_relative_gain.

Source code in src/datajudge/requirements.py

def add_n_rows_min_gain_constraint(
    self,
    constant_min_relative_gain: float | None = None,
    date_range_gain_deviation: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    r"""Assert that the number of rows hasn't grown less than expected.

    In particular, assert that

    $$n^{rows}_1 \geq n^{rows}_2 \cdot (1 + \text{cmrg})$$

    See readme for more information on ``constant_min_relative_gain``.
    """
    min_relative_gain_getter = self._get_deviation_getter(
        constant_min_relative_gain, date_range_gain_deviation
    )
    ref = DataReference(self._data_source, condition=condition1)
    ref2 = DataReference(self._data_source2, condition=condition2)
    self._constraints.append(
        nrows_constraints.NRowsMinGain(
            ref,
            ref2,
            min_relative_gain_getter,
            name=name,
            cache_size=cache_size,
        )
    )

add_n_uniques_equality_constraint

add_n_uniques_equality_constraint(columns1: list[str] | None, columns2: list[str] | None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_n_uniques_equality_constraint(
    self,
    columns1: list[str] | None,
    columns2: list[str] | None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        uniques_constraints.NUniquesEquality(
            ref, ref2=ref2, name=name, cache_size=cache_size
        )
    )

add_n_uniques_max_gain_constraint

add_n_uniques_max_gain_constraint(columns1: list[str] | None, columns2: list[str] | None, constant_max_relative_gain: float | None = None, date_range_gain_deviation: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the number of uniques hasn't grown by too much.

In particular, assert that

\[n^{uniques}_1 \leq n^{uniques}_2 \cdot (1 - \text{cmrg})\]

The number of uniques in first table are defined based on columns1, the number of uniques in second table are defined based on columns2.

See readme for more information on constant_max_relative_gain.

Source code in src/datajudge/requirements.py

def add_n_uniques_max_gain_constraint(
    self,
    columns1: list[str] | None,
    columns2: list[str] | None,
    constant_max_relative_gain: float | None = None,
    date_range_gain_deviation: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    r"""Assert that the number of uniques hasn't grown by too much.

    In particular, assert that

    $$n^{uniques}_1 \leq n^{uniques}_2 \cdot (1 - \text{cmrg})$$

    The number of uniques in first table are defined based on ``columns1``, the
    number of uniques in second table are defined based on ``columns2``.

    See readme for more information on ``constant_max_relative_gain``.
    """
    max_relative_gain_getter = self._get_deviation_getter(
        constant_max_relative_gain, date_range_gain_deviation
    )
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        uniques_constraints.NUniquesMaxGain(
            ref,
            ref2,
            max_relative_gain_getter,
            name=name,
            cache_size=cache_size,
        )
    )

add_n_uniques_max_loss_constraint

add_n_uniques_max_loss_constraint(columns1: list[str] | None, columns2: list[str] | None, constant_max_relative_loss: float | None = None, date_range_loss_deviation: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the number of uniques hasn't decreased too much.

In particular, assert that

\[n^{uniques}_1 \geq n^{uniques}_2 \cdot (1 - \text{cmrl})\]

The number of uniques in first table are defined based on columns1, the number of uniques in second table are defined based on columns2.

See readme for more information on constant_max_relative_loss.

Source code in src/datajudge/requirements.py

def add_n_uniques_max_loss_constraint(
    self,
    columns1: list[str] | None,
    columns2: list[str] | None,
    constant_max_relative_loss: float | None = None,
    date_range_loss_deviation: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    r"""Assert that the number of uniques hasn't decreased too much.

    In particular, assert that

    $$n^{uniques}_1 \geq n^{uniques}_2 \cdot (1 - \text{cmrl})$$

    The number of uniques in first table are defined based on ``columns1``, the
    number of uniques in second table are defined based on ``columns2``.

    See readme for more information on ``constant_max_relative_loss``.
    """
    max_relative_loss_getter = self._get_deviation_getter(
        constant_max_relative_loss, date_range_loss_deviation
    )
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        uniques_constraints.NUniquesMaxLoss(
            ref,
            ref2,
            max_relative_loss_getter,
            name=name,
            cache_size=cache_size,
        )
    )

add_numeric_max_constraint

add_numeric_max_constraint(column1: str, column2: str, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_numeric_max_constraint(
    self,
    column1: str,
    column2: str,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        numeric_constraints.NumericMax(
            ref, ref2=ref2, name=name, cache_size=cache_size
        )
    )

add_numeric_mean_constraint

add_numeric_mean_constraint(column1: str, column2: str, max_absolute_deviation: float, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_numeric_mean_constraint(
    self,
    column1: str,
    column2: str,
    max_absolute_deviation: float,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        numeric_constraints.NumericMean(
            ref,
            max_absolute_deviation,
            ref2=ref2,
            name=name,
            cache_size=cache_size,
        )
    )

add_numeric_min_constraint

add_numeric_min_constraint(column1: str, column2: str, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_numeric_min_constraint(
    self,
    column1: str,
    column2: str,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        numeric_constraints.NumericMin(
            ref, ref2=ref2, name=name, cache_size=cache_size
        )
    )

add_numeric_percentile_constraint

add_numeric_percentile_constraint(column1: str, column2: str, percentage: float, max_absolute_deviation: float | None = None, max_relative_deviation: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the percentage-th percentile is approximately equal.

The percentile is defined as the smallest value present in column1 / column2 for which percentage % of the values in column1 / column2 are less or equal. NULL values are ignored.

Hence, if percentage is less than the inverse of the number of non-NULL rows, None is received as the percentage-th percentile.

percentage is expected to be provided in percent. The median, for example, would correspond to percentage=50.

At least one of max_absolute_deviation and max_relative_deviation must be provided.

Source code in src/datajudge/requirements.py

def add_numeric_percentile_constraint(
    self,
    column1: str,
    column2: str,
    percentage: float,
    max_absolute_deviation: float | None = None,
    max_relative_deviation: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Assert that the ``percentage``-th percentile is approximately equal.

    The percentile is defined as the smallest value present in ``column1`` / ``column2``
    for which ``percentage`` % of the values in ``column1`` / ``column2`` are
    less or equal. ``NULL`` values are ignored.

    Hence, if ``percentage`` is less than the inverse of the number of non-``NULL``
    rows, ``None`` is received as the ``percentage``-th percentile.

    ``percentage`` is expected to be provided in percent. The median, for example,
    would correspond to ``percentage=50``.

    At least one of ``max_absolute_deviation`` and ``max_relative_deviation`` must
    be provided.
    """
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        numeric_constraints.NumericPercentile(
            ref,
            percentage=percentage,
            max_absolute_deviation=max_absolute_deviation,
            max_relative_deviation=max_relative_deviation,
            ref2=ref2,
            name=name,
            cache_size=cache_size,
        )
    )

add_row_equality_constraint

add_row_equality_constraint(columns1: list[str] | None, columns2: list[str] | None, max_missing_fraction: float, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

At most max_missing_fraction of rows in T1 and T2 are absent in either.

In other words

\[\frac{|T1 - T2| + |T2 - T1|}{|T1 \cup T2|} \leq \text{mmf}\]

Rows from T1 are indexed in columns1, rows from T2 are indexed in columns2.

Source code in src/datajudge/requirements.py

def add_row_equality_constraint(
    self,
    columns1: list[str] | None,
    columns2: list[str] | None,
    max_missing_fraction: float,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """At most ``max_missing_fraction`` of rows in T1 and T2 are absent in either.

    In other words

    $$\\frac{|T1 - T2| + |T2 - T1|}{|T1 \\cup T2|} \\leq \\text{mmf}$$

    Rows from T1 are indexed in ``columns1``, rows from T2 are indexed in ``columns2``.
    """  # noqa: D301
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        row_constraints.RowEquality(
            ref,
            ref2,
            lambda engine: max_missing_fraction,
            name=name,
            cache_size=cache_size,
        )
    )

add_row_matching_equality_constraint

add_row_matching_equality_constraint(matching_columns1: list[str], matching_columns2: list[str], comparison_columns1: list[str], comparison_columns2: list[str], max_missing_fraction: float, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Match tables in matching_columns, compare for equality in comparison_columns.

This constraint is similar to the nature of the RowEquality constraint. Just as the latter, this constraint divides the cardinality of an intersection by the cardinality of a union. The difference lies in how the set are created. While RowEquality considers all rows of both tables, indexed in columns, RowMatchingEquality considers only rows in both tables having values in matching_columns present in both tables. At most max_missing_fraction of such rows can be missing in the intersection.

Alternatively, this can be thought of as counting mismatches in comparison_columns after performing an inner join on matching_columns.

Source code in src/datajudge/requirements.py

def add_row_matching_equality_constraint(
    self,
    matching_columns1: list[str],
    matching_columns2: list[str],
    comparison_columns1: list[str],
    comparison_columns2: list[str],
    max_missing_fraction: float,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Match tables in matching_columns, compare for equality in comparison_columns.

    This constraint is similar to the nature of the ``RowEquality``
    constraint. Just as the latter, this constraint divides the
    cardinality of an intersection by the cardinality of a union.
    The difference lies in how the set are created. While ``RowEquality``
    considers all rows of both tables, indexed in columns,
    ``RowMatchingEquality`` considers only rows in both tables having values
    in ``matching_columns`` present in both tables. At most ``max_missing_fraction``
    of such rows can be missing in the intersection.

    Alternatively, this can be thought of as counting mismatches in
    ``comparison_columns`` after performing an inner join on ``matching_columns``.
    """
    ref = DataReference(
        self._data_source, matching_columns1 + comparison_columns1, condition1
    )
    ref2 = DataReference(
        self._data_source2, matching_columns2 + comparison_columns2, condition2
    )
    self._constraints.append(
        row_constraints.RowMatchingEquality(
            ref,
            ref2,
            matching_columns1,
            matching_columns2,
            comparison_columns1,
            comparison_columns2,
            lambda engine: max_missing_fraction,
            name=name,
            cache_size=cache_size,
        )
    )

add_row_subset_constraint

add_row_subset_constraint(columns1: list[str] | None, columns2: list[str] | None, constant_max_missing_fraction: float | None, date_range_loss_fraction: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

At most max_missing_fraction of rows in T1 are not in T2.

In other words, :math:\frac{|T1-T2|}{|T1|} \leq max_missing_fraction. Rows from T1 are indexed in columns1, rows from T2 are indexed in columns2.

In particular, the operation |T1-T2| relies on a sql EXCEPT statement. In contrast to EXCEPT ALL, this should lead to a set subtraction instead of a multiset subtraction. In other words, duplicates in T1 are treated as single occurrences.

Source code in src/datajudge/requirements.py

def add_row_subset_constraint(
    self,
    columns1: list[str] | None,
    columns2: list[str] | None,
    constant_max_missing_fraction: float | None,
    date_range_loss_fraction: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """At most ``max_missing_fraction`` of rows in T1 are not in T2.

    In other words,
    :math:`\\frac{|T1-T2|}{|T1|} \\leq` ``max_missing_fraction``.
    Rows from T1 are indexed in columns1, rows from T2 are indexed in ``columns2``.

    In particular, the operation ``|T1-T2|`` relies on a sql ``EXCEPT`` statement. In
    contrast to ``EXCEPT ALL``, this should lead to a set subtraction instead of
    a multiset subtraction. In other words, duplicates in T1 are treated as
    single occurrences.
    """  # noqa: D301
    max_missing_fraction_getter = self._get_deviation_getter(
        constant_max_missing_fraction, date_range_loss_fraction
    )
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        row_constraints.RowSubset(
            ref,
            ref2,
            max_missing_fraction_getter,
            name=name,
            cache_size=cache_size,
        )
    )

add_row_superset_constraint

add_row_superset_constraint(columns1: list[str] | None, columns2: list[str] | None, constant_max_missing_fraction: float, date_range_loss_fraction: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

At most max_missing_fraction of rows in T2 are not in T1.

In other words, \(\frac{|T2-T1|}{|T2|} \leq\) max_missing_fraction. Rows from T1 are indexed in columns1, rows from T2 are indexed in columns2.

Source code in src/datajudge/requirements.py

def add_row_superset_constraint(
    self,
    columns1: list[str] | None,
    columns2: list[str] | None,
    constant_max_missing_fraction: float,
    date_range_loss_fraction: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """At most ``max_missing_fraction`` of rows in T2 are not in T1.

    In other words, $\\frac{|T2-T1|}{|T2|} \\leq$ ``max_missing_fraction``.
    Rows from T1 are indexed in ``columns1``, rows from T2 are indexed in
    ``columns2``.
    """  # noqa: D301
    max_missing_fraction_getter = self._get_deviation_getter(
        constant_max_missing_fraction, date_range_loss_fraction
    )
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        row_constraints.RowSuperset(
            ref,
            ref2,
            max_missing_fraction_getter,
            name=name,
            cache_size=cache_size,
        )
    )

add_uniques_equality_constraint

add_uniques_equality_constraint(columns1: list[str], columns2: list[str], filter_func: Callable[[list[_T]], list[_T]] | None = None, map_func: Callable[[_T], _T] | None = None, reduce_func: Callable[[Collection], Collection] | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Check if the data's unique values in given columns are equal.

The UniquesEquality constraint asserts if the values contained in a column of a DataSource's columns, are strictly the ones of another DataSource's columns.

Null values in the columns columns are ignored. To assert the non-existence of them use the add_null_absence_constraint helper method for WithinRequirement. By default, the null filtering does not trigger if multiple columns are fetched at once. It can be configured in more detail by supplying a custom filter_func function. Some exemplary implementations are available as filternull_element, filternull_never, filternull_element_or_tuple_all, filternull_element_or_tuple_any. Passing None as the argument is equivalent to filternull_element but triggers a warning. The current default of filternull_element Cause (possibly often unintended) changes in behavior when the users adds a second column (filtering no longer can trigger at all). The default will be changed to filternull_element_or_tuple_all in future versions. To silence the warning, set filter_func explicitly..

See Uniques for further parameter details on map_func, reduce_func, and output_processors.

Source code in src/datajudge/requirements.py

def add_uniques_equality_constraint(
    self,
    columns1: list[str],
    columns2: list[str],
    filter_func: Callable[[list[_T]], list[_T]] | None = None,
    map_func: Callable[[_T], _T] | None = None,
    reduce_func: Callable[[Collection], Collection] | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Check if the data's unique values in given columns are equal.

    The ``UniquesEquality`` constraint asserts if the values contained in a column
    of a ``DataSource``'s columns, are strictly the ones of another ``DataSource``'s
    columns.

    Null values in the columns ``columns`` are ignored. To assert the non-existence of them use
    the [`add_null_absence_constraint`][datajudge.requirements.WithinRequirement.add_null_absence_constraint] helper method
    for ``WithinRequirement``.
    By default, the null filtering does not trigger if multiple columns are fetched at once.
    It can be configured in more detail by supplying a custom ``filter_func`` function.
    Some exemplary implementations are available as [`filternull_element`][datajudge.utils.filternull_element],
    [`filternull_never`][datajudge.utils.filternull_never], [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all],
    [`filternull_element_or_tuple_any`][datajudge.utils.filternull_element_or_tuple_any].
    Passing ``None`` as the argument is equivalent to [`filternull_element`][datajudge.utils.filternull_element] but triggers a warning.
    The current default of [`filternull_element`][datajudge.utils.filternull_element]
    Cause (possibly often unintended) changes in behavior when the users adds a second column
    (filtering no longer can trigger at all).
    The default will be changed to [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all] in future versions.
    To silence the warning, set ``filter_func`` explicitly..

    See [`Uniques`][datajudge.constraints.uniques.Uniques] for further parameter details on ``map_func``,
    ``reduce_func``, and ``output_processors``.
    """
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        uniques_constraints.UniquesEquality(
            ref,
            ref2=ref2,
            filter_func=filter_func,
            map_func=map_func,
            reduce_func=reduce_func,
            output_processors=output_processors,
            name=name,
            cache_size=cache_size,
        )
    )

add_uniques_subset_constraint

add_uniques_subset_constraint(columns1: list[str], columns2: list[str], max_relative_violations: float = 0, filter_func: Callable[[list[_T]], list[_T]] | None = None, compare_distinct: bool = False, map_func: Callable[[_T], _T] | None = None, reduce_func: Callable[[Collection], Collection] | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, cache_size=None) -> None

Check if the given columns's unique values in are contained in reference data.

The UniquesSubset constraint asserts if the values contained in given column of a DataSource are part of the unique values of given columns of another DataSource.

Null values in the columns columns are ignored. To assert the non-existence of them use the add_null_absence_constraint helper method for WithinRequirement. By default, the null filtering does not trigger if multiple columns are fetched at once. It can be configured in more detail by supplying a custom filter_func function. Some exemplary implementations are available as filternull_element, filternull_never, filternull_element_or_tuple_all, filternull_element_or_tuple_any. Passing None as the argument is equivalent to filternull_element but triggers a warning. The current default of filternull_element Cause (possibly often unintended) changes in behavior when the users adds a second column (filtering no longer can trigger at all). The default will be changed to filternull_element_or_tuple_all in future versions. To silence the warning, set filter_func explicitly. max_relative_violations indicates what fraction of rows of the given table may have values not included in the reference set of unique values. Please note that UniquesSubset and UniquesSuperset are not symmetrical in this regard.

By default, the number of occurrences affects the computed fraction of violations. To disable this weighting, set compare_distinct=True. This argument does not have an effect on the test results for other Uniques constraints, or if max_relative_violations is 0.

See Uniques for further details on map_func, reduce_func, and output_processors.

Source code in src/datajudge/requirements.py

def add_uniques_subset_constraint(
    self,
    columns1: list[str],
    columns2: list[str],
    max_relative_violations: float = 0,
    filter_func: Callable[[list[_T]], list[_T]] | None = None,
    compare_distinct: bool = False,
    map_func: Callable[[_T], _T] | None = None,
    reduce_func: Callable[[Collection], Collection] | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    cache_size=None,
) -> None:
    """Check if the given columns's unique values in are contained in reference data.

    The ``UniquesSubset`` constraint asserts if the values contained in given column of
    a ``DataSource`` are part of the unique values of given columns of another
    ``DataSource``.

    Null values in the columns ``columns`` are ignored. To assert the non-existence of them use
    the [`add_null_absence_constraint`][datajudge.requirements.WithinRequirement.add_null_absence_constraint] helper method
    for ``WithinRequirement``.
    By default, the null filtering does not trigger if multiple columns are fetched at once.
    It can be configured in more detail by supplying a custom ``filter_func`` function.
    Some exemplary implementations are available as [`filternull_element`][datajudge.utils.filternull_element],
    [`filternull_never`][datajudge.utils.filternull_never], [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all],
    [`filternull_element_or_tuple_any`][datajudge.utils.filternull_element_or_tuple_any].
    Passing ``None`` as the argument is equivalent to [`filternull_element`][datajudge.utils.filternull_element] but triggers a warning.
    The current default of [`filternull_element`][datajudge.utils.filternull_element]
    Cause (possibly often unintended) changes in behavior when the users adds a second column
    (filtering no longer can trigger at all).
    The default will be changed to [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all] in future versions.
    To silence the warning, set ``filter_func`` explicitly.
    ``max_relative_violations`` indicates what fraction of rows of the given table
    may have values not included in the reference set of unique values. Please note
    that ``UniquesSubset`` and ``UniquesSuperset`` are not symmetrical in this regard.

    By default, the number of occurrences affects the computed fraction of violations.
    To disable this weighting, set ``compare_distinct=True``.
    This argument does not have an effect on the test results for other [`Uniques`][datajudge.constraints.uniques.Uniques] constraints,
    or if ``max_relative_violations`` is 0.

    See [`Uniques`][datajudge.constraints.uniques.Uniques] for further details on ``map_func``, ``reduce_func``,
    and ``output_processors``.
    """
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        uniques_constraints.UniquesSubset(
            ref,
            ref2=ref2,
            max_relative_violations=max_relative_violations,
            compare_distinct=compare_distinct,
            filter_func=filter_func,
            map_func=map_func,
            reduce_func=reduce_func,
            output_processors=output_processors,
            name=name,
            cache_size=cache_size,
        )
    )

add_uniques_superset_constraint

add_uniques_superset_constraint(columns1: list[str], columns2: list[str], max_relative_violations: float = 0, filter_func: Callable[[list[_T]], list[_T]] | None = None, map_func: Callable[[_T], _T] | None = None, reduce_func: Callable[[Collection], Collection] | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, cache_size=None) -> None

Check if unique values of columns are contained in the reference data.

The UniquesSuperset constraint asserts that reference set of expected values, derived from the unique values in given columns of the reference DataSource, is contained in given columns of a DataSource.

Null values in the columns columns are ignored. To assert the non-existence of them use the add_null_absence_constraint helper method for WithinRequirement. By default, the null filtering does not trigger if multiple columns are fetched at once. It can be configured in more detail by supplying a custom filter_func function. Some exemplary implementations are available as filternull_element, filternull_never, filternull_element_or_tuple_all, filternull_element_or_tuple_any. Passing None as the argument is equivalent to filternull_element but triggers a warning. The current default of filternull_element Cause (possibly often unintended) changes in behavior when the users adds a second column (filtering no longer can trigger at all). The default will be changed to filternull_element_or_tuple_all in future versions. To silence the warning, set filter_func explicitly..

max_relative_violations indicates what fraction of unique values of the given DataSource are not represented in the reference set of unique values. Please note that UniquesSubset and UniquesSuperset are not symmetrical in this regard.

One use of this constraint is to test for consistency in columns with expected categorical values.

See Uniques for further details on map_func, reduce_func, and output_processors.

Source code in src/datajudge/requirements.py

def add_uniques_superset_constraint(
    self,
    columns1: list[str],
    columns2: list[str],
    max_relative_violations: float = 0,
    filter_func: Callable[[list[_T]], list[_T]] | None = None,
    map_func: Callable[[_T], _T] | None = None,
    reduce_func: Callable[[Collection], Collection] | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    cache_size=None,
) -> None:
    """Check if unique values of columns are contained in the reference data.

    The ``UniquesSuperset`` constraint asserts that reference set of expected values,
    derived from the unique values in given columns of the reference ``DataSource``,
    is contained in given columns of a ``DataSource``.

    Null values in the columns ``columns`` are ignored. To assert the non-existence of them use
    the [`add_null_absence_constraint`][datajudge.requirements.WithinRequirement.add_null_absence_constraint] helper method
    for ``WithinRequirement``.
    By default, the null filtering does not trigger if multiple columns are fetched at once.
    It can be configured in more detail by supplying a custom ``filter_func`` function.
    Some exemplary implementations are available as [`filternull_element`][datajudge.utils.filternull_element],
    [`filternull_never`][datajudge.utils.filternull_never], [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all],
    [`filternull_element_or_tuple_any`][datajudge.utils.filternull_element_or_tuple_any].
    Passing ``None`` as the argument is equivalent to [`filternull_element`][datajudge.utils.filternull_element] but triggers a warning.
    The current default of [`filternull_element`][datajudge.utils.filternull_element]
    Cause (possibly often unintended) changes in behavior when the users adds a second column
    (filtering no longer can trigger at all).
    The default will be changed to [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all] in future versions.
    To silence the warning, set ``filter_func`` explicitly..

    ``max_relative_violations`` indicates what fraction of unique values of the given
    ``DataSource`` are not represented in the reference set of unique values. Please
    note that ``UniquesSubset`` and ``UniquesSuperset`` are not symmetrical in this regard.

    One use of this constraint is to test for consistency in columns with expected
    categorical values.

    See [`Uniques`][datajudge.constraints.uniques.Uniques] for further details on ``map_func``, ``reduce_func``,
    and ``output_processors``.
    """
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        uniques_constraints.UniquesSuperset(
            ref,
            ref2=ref2,
            max_relative_violations=max_relative_violations,
            filter_func=filter_func,
            map_func=map_func,
            reduce_func=reduce_func,
            output_processors=output_processors,
            name=name,
            cache_size=cache_size,
        )
    )

add_varchar_max_length_constraint

add_varchar_max_length_constraint(column1: str, column2: str, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_varchar_max_length_constraint(
    self,
    column1: str,
    column2: str,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        varchar_constraints.VarCharMaxLength(
            ref, ref2=ref2, name=name, cache_size=cache_size
        )
    )

add_varchar_min_length_constraint

add_varchar_min_length_constraint(column1: str, column2: str, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_varchar_min_length_constraint(
    self,
    column1: str,
    column2: str,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        varchar_constraints.VarCharMinLength(
            ref, ref2=ref2, name=name, cache_size=cache_size
        )
    )

from_expressions classmethod

from_expressions(expression1, expression2, name1: str, name2: str, date_column: str | None = None, date_column2: str | None = None)

Create a BetweenTableRequirement based on sqlalchemy expressions.

Any sqlalchemy object implementing the alias method can be passed as an argument for the expression1 and expression2 parameters. This could, e.g. be a sqlalchemy.Table object or the result of a sqlalchemy.select invocation.

name1 and name2 will be used to represent the expressions in error messages, respectively.

Source code in src/datajudge/requirements.py

@classmethod
def from_expressions(
    cls,
    expression1,
    expression2,
    name1: str,
    name2: str,
    date_column: str | None = None,
    date_column2: str | None = None,
):
    """Create a ``BetweenTableRequirement`` based on sqlalchemy expressions.

    Any sqlalchemy object implementing the ``alias`` method can be passed as an
    argument for the ``expression1`` and ``expression2`` parameters. This could,
    e.g. be a ``sqlalchemy.Table`` object or the result of a ``sqlalchemy.select``
    invocation.

    ``name1`` and ``name2`` will be used to represent the expressions in error messages,
    respectively.
    """
    return cls(
        data_source=ExpressionDataSource(expression1, name1),
        data_source2=ExpressionDataSource(expression2, name2),
        date_column=date_column,
        date_column2=date_column2,
    )

from_raw_queries classmethod

from_raw_queries(query1: str, query2: str, name1: str, name2: str, columns1: list[str] | None = None, columns2: list[str] | None = None, date_column: str | None = None, date_column2: str | None = None)

Create a BetweenRequirement based on raw query strings.

The query1 and query2 parameters can be passed any query string returning rows, e.g. "SELECT * FROM myschema.mytable LIMIT 1337" or "SELECT id, name FROM table1 UNION SELECT id, name FROM table2".

name1 and name2 will be used to represent the queries in error messages, respectively.

If constraints rely on specific columns, these should be provided here via columns1 and columns2 respectively.

Source code in src/datajudge/requirements.py

@classmethod
def from_raw_queries(
    cls,
    query1: str,
    query2: str,
    name1: str,
    name2: str,
    columns1: list[str] | None = None,
    columns2: list[str] | None = None,
    date_column: str | None = None,
    date_column2: str | None = None,
):
    """Create a ``BetweenRequirement`` based on raw query strings.

    The ``query1`` and ``query2`` parameters can be passed any query string returning
    rows, e.g. ``"SELECT * FROM myschema.mytable LIMIT 1337"`` or
    ``"SELECT id, name FROM table1 UNION SELECT id, name FROM table2"``.

    ``name1`` and ``name2`` will be used to represent the queries in error messages,
    respectively.

    If constraints rely on specific columns, these should be provided here via
    ``columns1`` and ``columns2`` respectively.
    """
    return cls(
        data_source=RawQueryDataSource(query1, name1, columns=columns1),
        data_source2=RawQueryDataSource(query2, name2, columns=columns2),
        date_column=date_column,
        date_column2=date_column2,
    )

from_tables classmethod

from_tables(db_name1: str, schema_name1: str, table_name1: str, db_name2: str, schema_name2: str, table_name2: str, date_column: str | None = None, date_column2: str | None = None)

Create a BetweenRequirement based on a table.

Source code in src/datajudge/requirements.py

@classmethod
def from_tables(
    cls,
    db_name1: str,
    schema_name1: str,
    table_name1: str,
    db_name2: str,
    schema_name2: str,
    table_name2: str,
    date_column: str | None = None,
    date_column2: str | None = None,
):
    """Create a ``BetweenRequirement`` based on a table."""
    return cls(
        data_source=TableDataSource(
            db_name=db_name1,
            schema_name=schema_name1,
            table_name=table_name1,
        ),
        data_source2=TableDataSource(
            db_name=db_name2,
            schema_name=schema_name2,
            table_name=table_name2,
        ),
        date_column=date_column,
        date_column2=date_column2,
    )

Condition dataclass

Condition(raw_string: str | None = None, conditions: Sequence[Condition] | None = None, reduction_operator: str | None = None)

Condition allows for further narrowing down of a DataSource in a Constraint.

A Condition can be thought of as a filter, the content of a sql 'where' clause or a condition as known from probability theory.

While a DataSource is expressed more generally, one might be interested in testing properties of a specific part of said DataSource in light of a particular constraint. Hence using Condition allows for the reusage of a DataSource, in lieu of creating a new custom DataSource with the Condition implicitly built in.

A Condition can either be 'atomic', i.e. not further reducible to sub-conditions or 'composite', i.e. combining multiple subconditions. In the former case, it can be instantiated with help of the raw_string parameter, e.g. "col1 > 0". In the latter case, it can be instantiated with help of the conditions and reduction_operator parameters. reduction_operator allows for two values: "and" (logical conjunction) and "or" (logical disjunction). Note that composition of Condition supports arbitrary degrees of nesting.

conditions class-attribute instance-attribute

conditions: Sequence[Condition] | None = None

raw_string class-attribute instance-attribute

raw_string: str | None = None

reduction_operator class-attribute instance-attribute

reduction_operator: str | None = None

Constraint

Constraint(ref: DataReference, *, ref2: DataReference | None = None, ref_value: Any = None, name: str | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, cache_size=None)

Bases: ABC

Express a DataReference constraint against either another DataReference or a reference value.

Constraints against other DataReferences are typically referred to as 'between' constraints. Please use the the ref2 argument to instantiate such a constraint. Constraints against a fixed reference value are typically referred to as 'within' constraints. Please use the ref_value argument to instantiate such a constraint.

A constraint typically relies on the comparison of factual and target values. The former represent the key quantity of interest as seen in the database, the latter the key quantity of interest as expected a priori. Such a comparison is meant to be carried out in the test method.

In order to obtain such values, the retrieve method defines a mapping from DataReference, be it the DataReference of primary interest, ref, or a baseline DataReference, ref2, to value. If ref_value is already provided, usually no further mapping needs to be taken care of.

By default, retrieved arguments are cached indefinitely @lru_cache(maxsize=None). This can be controlled by setting the cache_size argument to a different value. 0 disables caching.

Source code in src/datajudge/constraints/base.py

def __init__(
    self,
    ref: DataReference,
    *,
    ref2: DataReference | None = None,
    ref_value: Any = None,
    name: str | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    cache_size=None,
):
    self._check_if_valid_between_or_within(ref2, ref_value)
    self._ref = ref
    self._ref2 = ref2
    self._ref_value = ref_value
    self.name = name
    self._factual_selections: _OptionalSelections = None
    self._target_selections: _OptionalSelections = None
    self._factual_queries: list[str] | None = None
    self._target_queries: list[str] | None = None

    if (output_processors is not None) and (
        not isinstance(output_processors, list)
    ):
        output_processors = [output_processors]
    self._output_processors = output_processors

    self._cache_size = cache_size
    self._setup_caching()

name instance-attribute

name = name

get_description

get_description() -> str

Source code in src/datajudge/constraints/base.py

def get_description(self) -> str:
    if self.name is not None:
        return self.name
    if self._ref2 is None:
        data_source_string = str(self._ref.data_source)
    else:
        data_source1_string = str(self._ref.data_source)
        data_source2_string = str(self._ref2.data_source)

        data_source1_substring, data_source2_substring = _uncommon_substrings(
            data_source1_string, data_source2_string
        )
        data_source_string = f"{data_source1_substring} | {data_source2_substring}"
    return self.__class__.__name__ + "::" + data_source_string

test

test(engine: Engine) -> TestResult

Source code in src/datajudge/constraints/base.py

def test(self, engine: sa.engine.Engine) -> TestResult:
    value_factual = self._get_factual_value(engine)
    value_target = self._get_target_value(engine)
    is_success, assertion_message = self._compare(value_factual, value_target)
    if is_success:
        return TestResult.success()

    factual_queries = None
    if self._factual_selections:
        factual_queries = [
            str(
                factual_selection.compile(
                    engine, compile_kwargs={"literal_binds": True}
                )
            )
            for factual_selection in self._factual_selections
        ]
    target_queries = None
    if self._target_selections:
        target_queries = [
            str(
                target_selection.compile(
                    engine, compile_kwargs={"literal_binds": True}
                )
            )
            for target_selection in self._target_selections
        ]
    return TestResult.failure(
        assertion_message,
        self.get_description(),
        factual_queries,
        target_queries,
    )

DataSource

Bases: ABC

Requirement

Requirement()

Bases: ABC, MutableSequence

Source code in src/datajudge/requirements.py

def __init__(self):
    self._constraints: list[Constraint] = []
    self._data_source: DataSource

insert

insert(index: int, value: Constraint) -> None

Source code in src/datajudge/requirements.py

def insert(self, index: int, value: Constraint) -> None:
    self._constraints.insert(index, value)

test

test(engine) -> list[TestResult]

Source code in src/datajudge/requirements.py

def test(self, engine) -> list[TestResult]:
    return [constraint.test(engine) for constraint in self]

WithinRequirement

WithinRequirement(data_source: DataSource)

Bases: Requirement

Source code in src/datajudge/requirements.py

def __init__(self, data_source: DataSource):
    self._data_source = data_source
    super().__init__()

add_categorical_bound_constraint

add_categorical_bound_constraint(columns: list[str], distribution: dict[_T, tuple[float, float]], default_bounds: tuple[float, float] = (0, 0), max_relative_violations: float = 0, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Check if the distribution of unique values in columns falls within the specified minimum and maximum bounds.

The CategoricalBoundConstraint is added to ensure the distribution of unique values in the specified columns of a DataSource falls within the given minimum and maximum bounds defined in the distribution parameter.

PARAMETER	DESCRIPTION
columns	A list of column names from the DataSource to apply the constraint on. TYPE: list[str]
distribution	A dictionary where keys represent unique values and the corresponding tuple values represent the minimum and maximum allowed proportions of the respective unique value in the columns. TYPE: dict[_T, tuple[float, float]]
default_bounds	A tuple specifying the minimum and maximum allowed proportions for all elements not mentioned in the distribution. By default, it's set to (0, 0), which means all elements not present in distribution will cause a constraint failure. TYPE: tuple[float, float] DEFAULT: (0, 0)
max_relative_violations	A tolerance threshold (0 to 1) for the proportion of elements in the data that can violate the bound constraints without triggering the constraint violation. TYPE: float DEFAULT: 0
condition	An optional parameter to specify a Condition object to filter the data before applying the constraint. TYPE: Condition | None DEFAULT: None
name	An optional parameter to provide a custom name for the constraint. TYPE: str | None DEFAULT: None
cache_size	TODO DEFAULT: None

Example:

This method can be used to test for consistency in columns with expected categorical values or ensure that the distribution of values in a column adheres to a certain criterion.

Usage:

requirement = WithinRequirement(data_source)
requirement.add_categorical_bound_constraint(
    columns=['column_name'],
    distribution={'A': (0.2, 0.3), 'B': (0.4, 0.6), 'C': (0.1, 0.2)},
    max_relative_violations=0.05,
    name='custom_name'
)

Source code in src/datajudge/requirements.py

def add_categorical_bound_constraint(
    self,
    columns: list[str],
    distribution: dict[_T, tuple[float, float]],
    default_bounds: tuple[float, float] = (0, 0),
    max_relative_violations: float = 0,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Check if the distribution of unique values in columns falls within the specified minimum and maximum bounds.

    The ``CategoricalBoundConstraint`` is added to ensure the distribution of unique values
    in the specified columns of a ``DataSource`` falls within the given minimum and maximum
    bounds defined in the ``distribution`` parameter.

    Parameters
    ----------
    columns:
        A list of column names from the `DataSource` to apply the constraint on.
    distribution:
        A dictionary where keys represent unique values and the corresponding
        tuple values represent the minimum and maximum allowed proportions of the respective
        unique value in the columns.
    default_bounds:
        A tuple specifying the minimum and maximum allowed proportions for all
        elements not mentioned in the distribution. By default, it's set to (0, 0), which means
        all elements not present in `distribution` will cause a constraint failure.
    max_relative_violations:
        A tolerance threshold (0 to 1) for the proportion of elements in the data that can violate the
        bound constraints without triggering the constraint violation.
    condition:
        An optional parameter to specify a `Condition` object to filter the data
        before applying the constraint.
    name:
        An optional parameter to provide a custom name for the constraint.
    cache_size:
        TODO

    Example:
    -------
    This method can be used to test for consistency in columns with expected categorical
    values or ensure that the distribution of values in a column adheres to a certain
    criterion.

    Usage:

    ```
    requirement = WithinRequirement(data_source)
    requirement.add_categorical_bound_constraint(
        columns=['column_name'],
        distribution={'A': (0.2, 0.3), 'B': (0.4, 0.6), 'C': (0.1, 0.2)},
        max_relative_violations=0.05,
        name='custom_name'
    )
    ```
    """
    ref = DataReference(self._data_source, columns, condition)
    self._constraints.append(
        uniques_constraints.CategoricalBoundConstraint(
            ref,
            distribution=distribution,
            default_bounds=default_bounds,
            max_relative_violations=max_relative_violations,
            name=name,
            cache_size=cache_size,
        )
    )

add_column_existence_constraint

add_column_existence_constraint(columns: list[str], name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_column_existence_constraint(
    self, columns: list[str], name: str | None = None, cache_size=None
) -> None:
    # Note that columns are not meant to be part of the reference.
    ref = DataReference(self._data_source)
    self._constraints.append(
        column_constraints.ColumnExistence(ref, columns, cache_size=cache_size)
    )

add_column_type_constraint

add_column_type_constraint(column: str, column_type: str | TypeEngine, name: str | None = None, cache_size=None) -> None

Check if a column type matches the expected column_type.

The column_type can be provided as a string (backend-specific type name), a backend-specific SQLAlchemy type, or a SQLAlchemy's generic type.

If SQLAlchemy's generic types are used, the check is performed using isinstance, which means that the actual type can also be a subclass of the target type. For more information on SQLAlchemy's generic types, see https://docs.sqlalchemy.org/en/20/core/type_basics.html

PARAMETER	DESCRIPTION
column	The name of the column to which the constraint will be applied. TYPE: str
column_type	The expected type of the column. This can be a string, a backend-specific SQLAlchemy type, or a generic SQLAlchemy type. TYPE: str | TypeEngine
name	An optional name for the constraint. If not provided, a name will be generated automatically. TYPE: str | None DEFAULT: None

Source code in src/datajudge/requirements.py

def add_column_type_constraint(
    self,
    column: str,
    column_type: str | sa.types.TypeEngine,
    name: str | None = None,
    cache_size=None,
) -> None:
    """
    Check if a column type matches the expected column_type.

    The ``column_type`` can be provided as a string (backend-specific type name), a backend-specific SQLAlchemy type, or a SQLAlchemy's generic type.

    If SQLAlchemy's generic types are used, the check is performed using ``isinstance``, which means that the actual type can also be a subclass of the target type.
    For more information on SQLAlchemy's generic types, see https://docs.sqlalchemy.org/en/20/core/type_basics.html

    Parameters
    ----------
    column : str
        The name of the column to which the constraint will be applied.

    column_type : str | sa.types.TypeEngine
        The expected type of the column. This can be a string, a backend-specific SQLAlchemy type, or a generic SQLAlchemy type.

    name : str | None
        An optional name for the constraint. If not provided, a name will be generated automatically.
    """
    ref = DataReference(self._data_source, [column])
    self._constraints.append(
        column_constraints.ColumnType(
            ref,
            column_type=column_type,
            name=name,
            cache_size=cache_size,
        )
    )

add_date_between_constraint

add_date_between_constraint(column: str, lower_bound: str, upper_bound: str, min_fraction: float, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Use string format: lower_bound="'20121230'".

Source code in src/datajudge/requirements.py

def add_date_between_constraint(
    self,
    column: str,
    lower_bound: str,
    upper_bound: str,
    min_fraction: float,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Use string format: ``lower_bound="'20121230'"``."""
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        date_constraints.DateBetween(
            ref,
            min_fraction,
            lower_bound,
            upper_bound,
            cache_size=cache_size,
        )
    )

add_date_max_constraint

add_date_max_constraint(column: str, max_value: str, use_upper_bound_reference: bool = True, column_type: str = 'date', condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Ensure all dates to be superior than max_value.

Use string format: max_value="'20121230'"

For more information on column_type values, see add_column_type_constraint.

If use_upper_bound_reference is True, the maximum date in column has to be smaller or equal to max_value. Otherwise the maximum date in column has to be greater or equal to max_value.

Source code in src/datajudge/requirements.py

def add_date_max_constraint(
    self,
    column: str,
    max_value: str,
    use_upper_bound_reference: bool = True,
    column_type: str = "date",
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Ensure all dates to be superior than ``max_value``.

    Use string format: ``max_value="'20121230'"``

    For more information on ``column_type`` values, see [`add_column_type_constraint`][datajudge.requirements.WithinRequirement.add_column_type_constraint].

    If ``use_upper_bound_reference`` is ``True``, the maximum date in ``column`` has to be smaller or
    equal to ``max_value``. Otherwise the maximum date in ``column`` has to be greater or equal
    to ``max_value``.
    """
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        date_constraints.DateMax(
            ref,
            max_value=max_value,
            use_upper_bound_reference=use_upper_bound_reference,
            column_type=column_type,
            name=name,
            cache_size=cache_size,
        )
    )

add_date_min_constraint

add_date_min_constraint(column: str, min_value: str, use_lower_bound_reference: bool = True, column_type: str = 'date', condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Ensure all dates to be superior than min_value.

Use string format: min_value="'20121230'".

For more information on column_type values, see add_column_type_constraint.

If use_lower_bound_reference, the min of the first table has to be greater or equal to min_value. If not use_upper_bound_reference, the min of the first table has to be smaller or equal to min_value.

Source code in src/datajudge/requirements.py

def add_date_min_constraint(
    self,
    column: str,
    min_value: str,
    use_lower_bound_reference: bool = True,
    column_type: str = "date",
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Ensure all dates to be superior than ``min_value``.

    Use string format: ``min_value="'20121230'"``.

    For more information on ``column_type`` values, see ``add_column_type_constraint``.

    If ``use_lower_bound_reference``, the min of the first table has to be
    greater or equal to ``min_value``.
    If not ``use_upper_bound_reference``, the min of the first table has to
    be smaller or equal to ``min_value``.
    """
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        date_constraints.DateMin(
            ref,
            min_value=min_value,
            use_lower_bound_reference=use_lower_bound_reference,
            column_type=column_type,
            name=name,
            cache_size=cache_size,
        )
    )

add_date_no_gap_constraint

add_date_no_gap_constraint(start_column: str, end_column: str, key_columns: list[str] | None = None, end_included: bool = True, max_relative_n_violations: float = 0, condition: Condition | None = None, name: str | None = None, cache_size=None)

Express that date range rows have no gap in-between them.

The table under inspection must consist of at least one but up to many key columns, identifying an entity. Additionally, a start_column and an end_column, indicating start and end dates, should be provided.

Neither of those columns should contain NULL values. Also, it should hold that for a given row, the value of end_column is strictly greater than the value of start_column.

Note that the value of start_column is expected to be included in each date range. By default, the value of end_column is expected to be included as well - this can however be changed by setting end_included to False.

A 'key' is a fixed set of values in key_columns and represents an entity of interest. A priori, a key is not a primary key, i.e., a key can have and often has several rows. Thereby, a key will often come with several date ranges.

If key_columns is None or [], all columns of the table will be considered as composing the key.

In order to express a tolerance for some violations of this gap property, use the max_relative_n_violations parameter. The latter expresses for what fraction of all key_values, at least one gap may exist.

For illustrative examples of this constraint, please refer to its test cases.

Source code in src/datajudge/requirements.py

def add_date_no_gap_constraint(
    self,
    start_column: str,
    end_column: str,
    key_columns: list[str] | None = None,
    end_included: bool = True,
    max_relative_n_violations: float = 0,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    """Express that date range rows have no gap in-between them.

    The table under inspection must consist of at least one but up to many key columns,
    identifying an entity. Additionally, a ``start_column`` and an ``end_column``,
    indicating start and end dates, should be provided.

    Neither of those columns should contain ``NULL`` values. Also, it should hold that
    for a given row, the value of ``end_column`` is strictly greater than the value of
    ``start_column``.

    Note that the value of ``start_column`` is expected to be included in each date range.
    By default, the value of ``end_column`` is expected to be included as well - this can
    however be changed by setting ``end_included`` to ``False``.

    A 'key' is a fixed set of values in ``key_columns`` and represents an entity of
    interest. A priori, a key is not a primary key, i.e., a key can have and often has
    several rows. Thereby, a key will often come with several date ranges.

    If ``key_columns`` is ``None`` or ``[]``, all columns of the table will be
    considered as composing the key.

    In order to express a tolerance for some violations of this gap property, use the
    ``max_relative_n_violations`` parameter. The latter expresses for what fraction
    of all key_values, at least one gap may exist.

    For illustrative examples of this constraint, please refer to its test cases.
    """
    relevant_columns = (
        ([start_column, end_column] + key_columns) if key_columns else []
    )
    ref = DataReference(self._data_source, relevant_columns, condition)
    self._constraints.append(
        date_constraints.DateNoGap(
            ref,
            key_columns=key_columns,
            start_columns=[start_column],
            end_columns=[end_column],
            max_relative_n_violations=max_relative_n_violations,
            legitimate_gap_size=1 if end_included else 0,
            name=name,
            cache_size=cache_size,
        )
    )

add_date_no_overlap_2d_constraint

add_date_no_overlap_2d_constraint(start_column1: str, end_column1: str, start_column2: str, end_column2: str, key_columns: list[str] | None = None, end_included: bool = True, max_relative_n_violations: float = 0, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Express that several date range rows do not overlap in two date dimensions.

The table under inspection must consist of at least one but up to many key columns, identifying an entity. Per date dimension, a start column (start_column1, start_column2) and end (end_column1, end_column2) column should be provided in order to define date ranges.

Date ranges in different date dimensions are expected to represent different kinds of dates. For example, let's say that a row in a table indicates an averag temperature forecast. start_column1 and end_column1 could the date span that was forecasted, e.g. the weather from next Saturday to next Sunday. end_column1 and end_column2 might indicate the timespan when this forceast was used, e.g. from the previous Monday to Wednesday.

Neither of those columns should contain NULL values. Also, the value of end_column_k should be strictly greater than the value of start_column_k.

Note that the values of start_column1 and start_column2 are expected to be included in each date range. By default, the values of end_column1 and end_column2 are expected to be included as well - this can however be changed by setting end_included to False.

A 'key' is a fixed set of values in key_columns and represents an entity of interest. A priori, a key is not a primary key, i.e., a key can have and often has several rows. Thereby, a key will often come with several date ranges.

Often, you might want the date ranges for a given key not to overlap.

If key_columns is None or [], all columns of the table will be considered as composing the key.

In order to express a tolerance for some violations of this non-overlapping property, use the max_relative_n_violations parameter. The latter expresses for what fraction of all key_values, at least one overlap may exist.

For illustrative examples of this constraint, please refer to its test cases.

Source code in src/datajudge/requirements.py

def add_date_no_overlap_2d_constraint(
    self,
    start_column1: str,
    end_column1: str,
    start_column2: str,
    end_column2: str,
    key_columns: list[str] | None = None,
    end_included: bool = True,
    max_relative_n_violations: float = 0,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Express that several date range rows do not overlap in two date dimensions.

    The table under inspection must consist of at least one but up to many key columns,
    identifying an entity. Per date dimension, a start column (``start_column1``, ``start_column2``)
    and end (``end_column1``, ``end_column2``) column should be provided in order to define
    date ranges.

    Date ranges in different date dimensions are expected to represent different kinds
    of dates. For example, let's say that a row in a table indicates an averag temperature
    forecast. ``start_column1`` and ``end_column1`` could the date span that was forecasted,
    e.g. the weather from next Saturday to next Sunday. ``end_column1`` and ``end_column2``
    might indicate the timespan when this forceast was used, e.g. from the
    previous Monday to Wednesday.

    Neither of those columns should contain ``NULL`` values. Also, the value of ``end_column_k``
    should be strictly greater than the value of ``start_column_k``.

    Note that the values of ``start_column1`` and ``start_column2`` are expected to be
    included in each date range. By default, the values of ``end_column1`` and
    ``end_column2`` are expected to be included as well - this can however be changed
    by setting ``end_included`` to ``False``.

    A 'key' is a fixed set of values in key_columns and represents an entity of
    interest. A priori, a key is not a primary key, i.e., a key can have and often has
    several rows. Thereby, a key will often come with several date ranges.

    Often, you might want the date ranges for a given key not to overlap.

    If key_columns is ``None`` or ``[]``, all columns of the table will be considered as
    composing the key.

    In order to express a tolerance for some violations of this non-overlapping property,
    use the ``max_relative_n_violations`` parameter. The latter expresses for what fraction
    of all key_values, at least one overlap may exist.

    For illustrative examples of this constraint, please refer to its test cases.
    """
    relevant_columns = (
        [start_column1, end_column1, start_column2, end_column2] + key_columns
        if key_columns
        else []
    )
    ref = DataReference(
        self._data_source,
        relevant_columns,
        condition,
    )
    self._constraints.append(
        date_constraints.DateNoOverlap2d(
            ref,
            key_columns=key_columns,
            start_columns=[start_column1, start_column2],
            end_columns=[end_column1, end_column2],
            end_included=end_included,
            max_relative_n_violations=max_relative_n_violations,
            name=name,
            cache_size=cache_size,
        )
    )

add_date_no_overlap_constraint

add_date_no_overlap_constraint(start_column: str, end_column: str, key_columns: list[str] | None = None, end_included: bool = True, max_relative_n_violations: float = 0, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Constraint expressing that several date range rows may not overlap.

The DataSource under inspection must consist of at least one but up to many key_columns, identifying an entity, a start_column and an end_column.

For a given row in this DataSource, start_column and end_column indicate a date range. Neither of those columns should contain NULL values. Also, it should hold that for a given row, the value of end_column is strictly greater than the value of start_column.

Note that the value of start_column is expected to be included in each date range. By default, the value of end_column is expected to be included as well - this can however be changed by setting end_included to False.

A 'key' is a fixed set of values in key_columns and represents an entity of interest. A priori, a key is not a primary key, i.e., a key can have and often has several rows. Thereby, a key will often come with several date ranges.

Often, you might want the date ranges for a given key not to overlap.

If key_columns is None or [], all columns of the table will be considered as composing the key.

In order to express a tolerance for some violations of this non-overlapping property, use the max_relative_n_violations parameter. The latter expresses for what fraction of all key values, at least one overlap may exist.

For illustrative examples of this constraint, please refer to its test cases.

Source code in src/datajudge/requirements.py

def add_date_no_overlap_constraint(
    self,
    start_column: str,
    end_column: str,
    key_columns: list[str] | None = None,
    end_included: bool = True,
    max_relative_n_violations: float = 0,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Constraint expressing that several date range rows may not overlap.

    The [`DataSource`][datajudge.DataSource] under inspection must consist of at least one but up
    to many ``key_columns``, identifying an entity, a ``start_column`` and an
    ``end_column``.

    For a given row in this [`DataSource`][datajudge.DataSource], ``start_column`` and ``end_column`` indicate a
    date range. Neither of those columns should contain NULL values. Also, it
    should hold that for a given row, the value of ``end_column`` is strictly greater
    than the value of ``start_column``.

    Note that the value of ``start_column`` is expected to be included in each date
    range. By default, the value of ``end_column`` is expected to be included as well -
    this can however be changed by setting ``end_included`` to ``False``.

    A 'key' is a fixed set of values in ``key_columns`` and represents an entity of
    interest. A priori, a key is not a primary key, i.e., a key can have and often
    has several rows. Thereby, a key will often come with several date ranges.

    Often, you might want the date ranges for a given key not to overlap.

    If ``key_columns`` is ``None`` or ``[]``, all columns of the table will be considered
    as composing the key.

    In order to express a tolerance for some violations of this non-overlapping
    property, use the ``max_relative_n_violations`` parameter. The latter expresses for
    what fraction of all key values, at least one overlap may exist.

    For illustrative examples of this constraint, please refer to its test cases.
    """
    relevant_columns = [start_column, end_column] + (
        key_columns if key_columns else []
    )
    ref = DataReference(self._data_source, relevant_columns, condition)
    self._constraints.append(
        date_constraints.DateNoOverlap(
            ref,
            key_columns=key_columns,
            start_columns=[start_column],
            end_columns=[end_column],
            end_included=end_included,
            max_relative_n_violations=max_relative_n_violations,
            name=name,
            cache_size=cache_size,
        )
    )

add_functional_dependency_constraint

add_functional_dependency_constraint(key_columns: list[str], value_columns: list[str], condition: Condition | None = None, name: str | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, cache_size=None)

Expresses a functional dependency, a constraint where the value_columns are uniquely determined by the key_columns.

This means that for each unique combination of values in the key_columns, there is exactly one corresponding combination of values in the value_columns.

The add_unique_constraint constraint is a special case of this constraint, where the key_columns are a primary key, and all other columns are included value_columns. This constraint allows for a more general definition of functional dependencies, where the key_columns are not necessarily a primary key.

An additional configuration option (for details see the analogous parameter in for Uniques-constraints) on how the output is sorted and how many counterexamples are shown is available as output_processors.

An additional configuration option (for details see the analogous parameter in for Uniques-constraints) on how the output is sorted and how many counterexamples are shown is available as output_processors.

For more information on functional dependencies, see https://en.wikipedia.org/wiki/Functional_dependency.

Source code in src/datajudge/requirements.py

def add_functional_dependency_constraint(
    self,
    key_columns: list[str],
    value_columns: list[str],
    condition: Condition | None = None,
    name: str | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    cache_size=None,
):
    """Expresses a functional dependency, a constraint where the `value_columns` are uniquely determined by the `key_columns`.

    This means that for each unique combination of values in the `key_columns`, there is exactly one corresponding combination of values in the `value_columns`.

    The ``add_unique_constraint`` constraint is a special case of this constraint, where the ``key_columns`` are a primary key,
    and all other columns are included ``value_columns``.
    This constraint allows for a more general definition of functional dependencies, where the ``key_columns`` are not necessarily a primary key.

    An additional configuration option (for details see the analogous parameter in for ``Uniques``-constraints)
    on how the output is sorted and how many counterexamples are shown is available as ``output_processors``.

    An additional configuration option (for details see the analogous parameter in for ``Uniques``-constraints)
    on how the output is sorted and how many counterexamples are shown is available as ``output_processors``.

    For more information on functional dependencies, see https://en.wikipedia.org/wiki/Functional_dependency.
    """
    relevant_columns = key_columns + value_columns
    ref = DataReference(self._data_source, relevant_columns, condition)
    self._constraints.append(
        miscs_constraints.FunctionalDependency(
            ref,
            key_columns=key_columns,
            output_processors=output_processors,
            name=name,
            cache_size=cache_size,
        )
    )

add_groupby_aggregation_constraint

add_groupby_aggregation_constraint(columns: Sequence[str], aggregation_column: str, start_value: int, tolerance: float = 0, condition: Condition | None = None, name: str | None = None, cache_size=None)

Check whether array aggregate corresponds to an integer range.

The DataSource is grouped by columns. Sql's array_agg function is then applied to the aggregate_column.

Since we expect aggregate_column to be a numeric column, this leads to a multiset of aggregated values. These values should correspond to the integers ranging from start_value to the cardinality of the multiset.

In order to allow for slight deviations from this pattern, tolerance expresses the fraction of all grouped-by rows, which may be incomplete ranges.

Source code in src/datajudge/requirements.py

def add_groupby_aggregation_constraint(
    self,
    columns: Sequence[str],
    aggregation_column: str,
    start_value: int,
    tolerance: float = 0,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    """Check whether array aggregate corresponds to an integer range.

    The ``DataSource`` is grouped by ``columns``. Sql's ``array_agg`` function is then
    applied to the ``aggregate_column``.

    Since we expect ``aggregate_column`` to be a numeric column, this leads to
    a multiset of aggregated values. These values should correspond to the integers
    ranging from ``start_value`` to the cardinality of the multiset.

    In order to allow for slight deviations from this pattern, ``tolerance`` expresses
    the fraction of all grouped-by rows, which may be incomplete ranges.
    """
    ref = DataReference(self._data_source, list(columns), condition)
    self._constraints.append(
        groupby_constraints.AggregateNumericRangeEquality(
            ref,
            aggregation_column=aggregation_column,
            tolerance=tolerance,
            start_value=start_value,
            name=name,
            cache_size=cache_size,
        )
    )

add_max_null_fraction_constraint

add_max_null_fraction_constraint(column: str, max_null_fraction: float, condition: Condition | None = None, name: str | None = None, cache_size=None)

Assert that column has less than a certain fraction of NULL values.

max_null_fraction is expected to lie within [0, 1].

Source code in src/datajudge/requirements.py

def add_max_null_fraction_constraint(
    self,
    column: str,
    max_null_fraction: float,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    """Assert that ``column`` has less than a certain fraction of ``NULL`` values.

    ``max_null_fraction`` is expected to lie within [0, 1].
    """
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        miscs_constraints.MaxNullFraction(
            ref,
            max_null_fraction=max_null_fraction,
            name=name,
            cache_size=cache_size,
        )
    )

add_n_rows_equality_constraint

add_n_rows_equality_constraint(n_rows: int, condition: Condition | None = None, name: str | None = None, cache_size=None)

Source code in src/datajudge/requirements.py

def add_n_rows_equality_constraint(
    self,
    n_rows: int,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    ref = DataReference(self._data_source, None, condition)
    self._constraints.append(
        nrows_constraints.NRowsEquality(
            ref, n_rows=n_rows, name=name, cache_size=cache_size
        )
    )

add_n_rows_max_constraint

add_n_rows_max_constraint(n_rows_max: int, condition: Condition | None = None, name: str | None = None, cache_size=None)

Source code in src/datajudge/requirements.py

def add_n_rows_max_constraint(
    self,
    n_rows_max: int,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    ref = DataReference(self._data_source, None, condition)
    self._constraints.append(
        nrows_constraints.NRowsMax(
            ref, n_rows=n_rows_max, name=name, cache_size=cache_size
        )
    )

add_n_rows_min_constraint

add_n_rows_min_constraint(n_rows_min: int, condition: Condition | None = None, name: str | None = None, cache_size=None)

Source code in src/datajudge/requirements.py

def add_n_rows_min_constraint(
    self,
    n_rows_min: int,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    ref = DataReference(self._data_source, None, condition)
    self._constraints.append(
        nrows_constraints.NRowsMin(
            ref, n_rows=n_rows_min, name=name, cache_size=cache_size
        )
    )

add_n_uniques_equality_constraint

add_n_uniques_equality_constraint(columns: list[str] | None, n_uniques: int, condition: Condition | None = None, name: str | None = None, cache_size=None)

Source code in src/datajudge/requirements.py

def add_n_uniques_equality_constraint(
    self,
    columns: list[str] | None,
    n_uniques: int,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    ref = DataReference(self._data_source, columns, condition)
    self._constraints.append(
        uniques_constraints.NUniquesEquality(
            ref, n_uniques=n_uniques, name=name, cache_size=cache_size
        )
    )

add_null_absence_constraint

add_null_absence_constraint(column: str, condition: Condition | None = None, name: str | None = None, cache_size=None)

Source code in src/datajudge/requirements.py

def add_null_absence_constraint(
    self,
    column: str,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        miscs_constraints.MaxNullFraction(
            ref, max_null_fraction=0, name=name, cache_size=cache_size
        )
    )

add_numeric_between_constraint

add_numeric_between_constraint(column: str, lower_bound: float, upper_bound: float, min_fraction: float, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the column's values lie between lower_bound and upper_bound.

Note that both bounds are inclusive.

Unless specified otherwise via the usage of a condition, NULL values will be considered in the denominator of min_fraction. NULL values will never be considered to lie in the interval [lower_bound, upper_bound].

Source code in src/datajudge/requirements.py

def add_numeric_between_constraint(
    self,
    column: str,
    lower_bound: float,
    upper_bound: float,
    min_fraction: float,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Assert that the column's values lie between ``lower_bound`` and ``upper_bound``.

    Note that both bounds are inclusive.

    Unless specified otherwise via the usage of a ``condition``, ``NULL`` values will
    be considered in the denominator of ``min_fraction``. ``NULL`` values will never be
    considered to lie in the interval [``lower_bound``, ``upper_bound``].
    """
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        numeric_constraints.NumericBetween(
            ref,
            min_fraction,
            lower_bound,
            upper_bound,
            name=name,
            cache_size=cache_size,
        )
    )

add_numeric_max_constraint

add_numeric_max_constraint(column: str, max_value: float, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

All values in column are less or equal max_value.

Source code in src/datajudge/requirements.py

def add_numeric_max_constraint(
    self,
    column: str,
    max_value: float,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """All values in ``column`` are less or equal ``max_value``."""
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        numeric_constraints.NumericMax(
            ref, max_value=max_value, name=name, cache_size=cache_size
        )
    )

add_numeric_mean_constraint

add_numeric_mean_constraint(column: str, mean_value: float, max_absolute_deviation: float, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert the mean of the column column deviates at most max_deviation from mean_value.

Source code in src/datajudge/requirements.py

def add_numeric_mean_constraint(
    self,
    column: str,
    mean_value: float,
    max_absolute_deviation: float,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Assert the mean of the column ``column`` deviates at most ``max_deviation`` from ``mean_value``."""
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        numeric_constraints.NumericMean(
            ref,
            max_absolute_deviation,
            mean_value=mean_value,
            name=name,
            cache_size=cache_size,
        )
    )

add_numeric_min_constraint

add_numeric_min_constraint(column: str, min_value: float, condition: Condition | None = None, cache_size=None) -> None

All values in column are greater or equal min_value.

Source code in src/datajudge/requirements.py

def add_numeric_min_constraint(
    self,
    column: str,
    min_value: float,
    condition: Condition | None = None,
    cache_size=None,
) -> None:
    """All values in ``column`` are greater or equal ``min_value``."""
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        numeric_constraints.NumericMin(
            ref, min_value=min_value, cache_size=cache_size
        )
    )

add_numeric_no_gap_constraint

add_numeric_no_gap_constraint(start_column: str, end_column: str, key_columns: list[str] | None = None, legitimate_gap_size: float = 0, max_relative_n_violations: float = 0, condition: Condition | None = None, name: str | None = None, cache_size=None)

Express that numeric interval rows have no gaps larger than some max value in-between them.

The table under inspection must consist of at least one but up to many key columns, identifying an entity. Additionally, a start_column and an end_column, indicating interval start and end values, should be provided.

Neither of those columns should contain NULL values. Also, it should hold that for a given row, the value of end_column is strictly greater than the value of start_column.

legitimate_gap_size is the maximum tollerated gap size between two intervals.

A 'key' is a fixed set of values in key_columns and represents an entity of interest. A priori, a key is not a primary key, i.e., a key can have and often has several rows. Thereby, a key will often come with several intervals.

If key_columns is None or [], all columns of the table will be considered as composing the key.

In order to express a tolerance for some violations of this gap property, use the max_relative_n_violations parameter. The latter expresses for what fraction of all key_values, at least one gap may exist.

For illustrative examples of this constraint, please refer to its test cases.

Source code in src/datajudge/requirements.py

def add_numeric_no_gap_constraint(
    self,
    start_column: str,
    end_column: str,
    key_columns: list[str] | None = None,
    legitimate_gap_size: float = 0,
    max_relative_n_violations: float = 0,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    """Express that numeric interval rows have no gaps larger than some max value in-between them.

    The table under inspection must consist of at least one but up to many key columns,
    identifying an entity. Additionally, a ``start_column`` and an ``end_column``,
    indicating interval start and end values, should be provided.

    Neither of those columns should contain ``NULL`` values. Also, it should hold that
    for a given row, the value of ``end_column`` is strictly greater than the value of
    ``start_column``.

    ``legitimate_gap_size`` is the maximum tollerated gap size between two intervals.

    A 'key' is a fixed set of values in ``key_columns`` and represents an entity of
    interest. A priori, a key is not a primary key, i.e., a key can have and often has
    several rows. Thereby, a key will often come with several intervals.

    If ``key_columns`` is ``None`` or ``[]``, all columns of the table will be
    considered as composing the key.

    In order to express a tolerance for some violations of this gap property, use the
    ``max_relative_n_violations`` parameter. The latter expresses for what fraction
    of all key_values, at least one gap may exist.

    For illustrative examples of this constraint, please refer to its test cases.
    """
    relevant_columns = (
        ([start_column, end_column] + key_columns) if key_columns else []
    )
    ref = DataReference(self._data_source, relevant_columns, condition)
    self._constraints.append(
        numeric_constraints.NumericNoGap(
            ref,
            key_columns=key_columns,
            start_columns=[start_column],
            end_columns=[end_column],
            legitimate_gap_size=legitimate_gap_size,
            max_relative_n_violations=max_relative_n_violations,
            name=name,
            cache_size=cache_size,
        )
    )

add_numeric_no_overlap_constraint

add_numeric_no_overlap_constraint(start_column: str, end_column: str, key_columns: list[str] | None = None, end_included: bool = True, max_relative_n_violations: float = 0, condition: Condition | None = None, name: str | None = None, cache_size=None)

Constraint expressing that several numeric interval rows may not overlap.

The DataSource under inspection must consist of at least one but up to many key_columns, identifying an entity, a start_column and an end_column.

For a given row in this DataSource, start_column and end_column indicate a numeric interval. Neither of those columns should contain NULL values. Also, it should hold that for a given row, the value of end_column is strictly greater than the value of start_column.

Note that the value of start_column is expected to be included in each interval. By default, the value of end_column is expected to be included as well - this can however be changed by setting end_included to False.

A 'key' is a fixed set of values in key_columns and represents an entity of interest. A priori, a key is not a primary key, i.e., a key can have and often has several rows. Thereby, a key will often come with several intervals.

Often, you might want the intervals for a given key not to overlap.

If key_columns is None or [], all columns of the table will be considered as composing the key.

In order to express a tolerance for some violations of this non-overlapping property, use the max_relative_n_violations parameter. The latter expresses for what fraction of all key values, at least one overlap may exist.

For illustrative examples of this constraint, please refer to its test cases.

Source code in src/datajudge/requirements.py

def add_numeric_no_overlap_constraint(
    self,
    start_column: str,
    end_column: str,
    key_columns: list[str] | None = None,
    end_included: bool = True,
    max_relative_n_violations: float = 0,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    """Constraint expressing that several numeric interval rows may not overlap.

    The ``DataSource`` under inspection must consist of at least one but up
    to many ``key_columns``, identifying an entity, a ``start_column`` and an
    ``end_column``.

    For a given row in this ``DataSource``, ``start_column`` and ``end_column`` indicate a
    numeric interval. Neither of those columns should contain NULL values. Also, it
    should hold that for a given row, the value of ``end_column`` is strictly greater
    than the value of ``start_column``.

    Note that the value of ``start_column`` is expected to be included in each interval.
    By default, the value of ``end_column`` is expected to be included as well -
    this can however be changed by setting ``end_included`` to ``False``.

    A 'key' is a fixed set of values in ``key_columns`` and represents an entity of
    interest. A priori, a key is not a primary key, i.e., a key can have and often
    has several rows. Thereby, a key will often come with several intervals.

    Often, you might want the intervals for a given key not to overlap.

    If ``key_columns`` is ``None`` or ``[]``, all columns of the table will be considered
    as composing the key.

    In order to express a tolerance for some violations of this non-overlapping
    property, use the ``max_relative_n_violations`` parameter. The latter expresses for
    what fraction of all key values, at least one overlap may exist.

    For illustrative examples of this constraint, please refer to its test cases.
    """
    relevant_columns = [start_column, end_column] + (
        key_columns if key_columns else []
    )
    ref = DataReference(self._data_source, relevant_columns, condition)
    self._constraints.append(
        numeric_constraints.NumericNoOverlap(
            ref,
            key_columns=key_columns,
            start_columns=[start_column],
            end_columns=[end_column],
            end_included=end_included,
            max_relative_n_violations=max_relative_n_violations,
            name=name,
            cache_size=cache_size,
        )
    )

add_numeric_percentile_constraint

add_numeric_percentile_constraint(column: str, percentage: float, expected_percentile: float, max_absolute_deviation: float | None = None, max_relative_deviation: float | None = None, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the percentage-th percentile is approximately expected_percentile.

The percentile is defined as the smallest value present in column for which percentage % of the values in column are less or equal. NULL values are ignored.

Hence, if percentage is less than the inverse of the number of non-NULL rows, None is received as the percentage -th percentile.

percentage is expected to be provided in percent. The median, for example, would correspond to percentage=50.

At least one of max_absolute_deviation and max_relative_deviation must be provided.

Source code in src/datajudge/requirements.py

def add_numeric_percentile_constraint(
    self,
    column: str,
    percentage: float,
    expected_percentile: float,
    max_absolute_deviation: float | None = None,
    max_relative_deviation: float | None = None,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Assert that the ``percentage``-th percentile is approximately ``expected_percentile``.

    The percentile is defined as the smallest value present in ``column`` for which
    ``percentage`` % of the values in ``column`` are less or equal. ``NULL`` values
    are ignored.

    Hence, if ``percentage`` is less than the inverse of the number of non-``NULL`` rows,
    ``None`` is received as the ``percentage`` -th percentile.

    ``percentage`` is expected to be provided in percent. The median, for example, would
    correspond to ``percentage=50``.

    At least one of ``max_absolute_deviation`` and ``max_relative_deviation`` must
    be provided.
    """
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        numeric_constraints.NumericPercentile(
            ref,
            percentage=percentage,
            expected_percentile=expected_percentile,
            max_absolute_deviation=max_absolute_deviation,
            max_relative_deviation=max_relative_deviation,
            name=name,
            cache_size=cache_size,
        )
    )

add_primary_key_definition_constraint

add_primary_key_definition_constraint(primary_keys: list[str], name: str | None = None, cache_size=None) -> None

Check that the primary key constraints in the database are exactly equal to the given column names.

Note that this doesn't actually check that the primary key values are unique across the table.

Source code in src/datajudge/requirements.py

def add_primary_key_definition_constraint(
    self,
    primary_keys: list[str],
    name: str | None = None,
    cache_size=None,
) -> None:
    """Check that the primary key constraints in the database are exactly equal to the given column names.

    Note that this doesn't actually check that the primary key values are unique across the table.
    """
    ref = DataReference(self._data_source)
    self._constraints.append(
        miscs_constraints.PrimaryKeyDefinition(
            ref, primary_keys, name=name, cache_size=cache_size
        )
    )

add_uniqueness_constraint

add_uniqueness_constraint(columns: list[str] | None = None, max_duplicate_fraction: float = 0, condition: Condition | None = None, max_absolute_n_duplicates: int = 0, infer_pk_columns: bool = False, name: str | None = None, cache_size=None) -> None

Columns should uniquely identify row.

Given a list of columns columns, validate the condition of a primary key, i.e. uniqueness of tuples in said columns. This constraint has a tolerance for inconsistencies, expressed via max_duplicate_fraction. The latter suggests that the number of uniques from said columns is larger or equal to 1 - max_duplicate_fraction times the number of rows.

If infer_pk_columns is True, columns will be retrieved from the primary keys. If columns is None and infer_pk_column is False, the fallback is validating that all rows in a table are unique.

Source code in src/datajudge/requirements.py

def add_uniqueness_constraint(
    self,
    columns: list[str] | None = None,
    max_duplicate_fraction: float = 0,
    condition: Condition | None = None,
    max_absolute_n_duplicates: int = 0,
    infer_pk_columns: bool = False,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Columns should uniquely identify row.

    Given a list of columns ``columns``, validate the condition of a primary key, i.e.
    uniqueness of tuples in said columns. This constraint has a tolerance
    for inconsistencies, expressed via ``max_duplicate_fraction``. The latter
    suggests that the number of uniques from said columns is larger or equal
    to ``1 - max_duplicate_fraction`` times the number of rows.

    If ``infer_pk_columns`` is ``True``, ``columns`` will be retrieved from the primary keys.
    If ``columns`` is ``None`` and ``infer_pk_column`` is ``False``, the fallback is
    validating that all rows in a table are unique.
    """
    ref = DataReference(self._data_source, columns, condition)
    self._constraints.append(
        miscs_constraints.Uniqueness(
            ref,
            max_duplicate_fraction=max_duplicate_fraction,
            max_absolute_n_duplicates=max_absolute_n_duplicates,
            infer_pk_columns=infer_pk_columns,
            name=name,
            cache_size=cache_size,
        )
    )

add_uniques_equality_constraint

add_uniques_equality_constraint(columns: list[str], uniques: Collection[_T], filter_func: Callable[[list[_T]], list[_T]] | None = None, map_func: Callable[[_T], _T] | None = None, reduce_func: Callable[[Collection], Collection] | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, condition: Condition | None = None, name: str | None = None, cache_size=None)

Check if the data's unique values are equal to a given set of values.

The UniquesEquality constraint asserts if the values contained in a column of a DataSource are strictly the ones of a reference set of expected values, specified via the uniques parameter.

Null values in the columns columns are ignored. To assert the non-existence of them use the add_null_absence_constraint helper method for WithinRequirement.

By default, the null filtering does not trigger if multiple columns are fetched at once. It can be configured in more detail by supplying a custom filter_func function. Some exemplary implementations are available as filternull_element, filternull_never, filternull_element_or_tuple_all, filternull_element_or_tuple_any. Passing None as the argument is equivalent to filternull_element but triggers a warning. The current default of filternull_element Cause (possibly often unintended) changes in behavior when the users adds a second column (filtering no longer can trigger at all). The default will be changed to filternull_element_or_tuple_all in future versions. To silence the warning, set filter_func explicitly.

See the Uniques class for further parameter details on map_func and reduce_func, and output_processors.

Source code in src/datajudge/requirements.py

def add_uniques_equality_constraint(
    self,
    columns: list[str],
    uniques: Collection[_T],
    filter_func: Callable[[list[_T]], list[_T]] | None = None,
    map_func: Callable[[_T], _T] | None = None,
    reduce_func: Callable[[Collection], Collection] | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    """Check if the data's unique values are equal to a given set of values.

    The `UniquesEquality` constraint asserts if the values contained in a column
    of a `DataSource` are strictly the ones of a reference set of expected values,
    specified via the `uniques` parameter.

    Null values in the columns `columns` are ignored. To assert the non-existence of them use
    the [`add_null_absence_constraint`][datajudge.requirements.WithinRequirement.add_null_absence_constraint] helper method
    for `WithinRequirement`.

    By default, the null filtering does not trigger if multiple columns are fetched at once.
    It can be configured in more detail by supplying a custom `filter_func` function.
    Some exemplary implementations are available as [`filternull_element`][datajudge.utils.filternull_element],
    [`filternull_never`][datajudge.utils.filternull_never], [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all],
    [`filternull_element_or_tuple_any`][datajudge.utils.filternull_element_or_tuple_any].
    Passing `None` as the argument is equivalent to [`filternull_element`][datajudge.utils.filternull_element] but triggers a warning.
    The current default of [`filternull_element`][datajudge.utils.filternull_element]
    Cause (possibly often unintended) changes in behavior when the users adds a second column
    (filtering no longer can trigger at all).
    The default will be changed to [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all] in future versions.
    To silence the warning, set `filter_func` explicitly.

    See the `Uniques` class for further parameter details on `map_func` and
    `reduce_func`, and `output_processors`.
    """
    ref = DataReference(self._data_source, columns, condition)
    self._constraints.append(
        uniques_constraints.UniquesEquality(
            ref,
            uniques=uniques,
            filter_func=filter_func,
            map_func=map_func,
            reduce_func=reduce_func,
            output_processors=output_processors,
            name=name,
            cache_size=cache_size,
        )
    )

add_uniques_subset_constraint

add_uniques_subset_constraint(columns: list[str], uniques: Collection[_T], max_relative_violations: float = 0, filter_func: Callable[[list[_T]], list[_T]] | None = None, compare_distinct: bool = False, map_func: Callable[[_T], _T] | None = None, reduce_func: Callable[[Collection], Collection] | None = None, condition: Condition | None = None, name: str | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, cache_size=None)

Check if the data's unique values are contained in a given set of values.

The UniquesSubset constraint asserts if the values contained in a column of a DataSource are part of a reference set of expected values, specified via uniques.

Null values in the columns columns are ignored. To assert the non-existence of them use the add_null_absence_constraint helper method for WithinRequirement. By default, the null filtering does not trigger if multiple columns are fetched at once. It can be configured in more detail by supplying a custom filter_func function. Some exemplary implementations are available as filternull_element, filternull_never, filternull_element_or_tuple_all, filternull_element_or_tuple_any. Passing None as the argument is equivalent to filternull_element but triggers a warning. The current default of filternull_element Cause (possibly often unintended) changes in behavior when the users adds a second column (filtering no longer can trigger at all). The default will be changed to filternull_element_or_tuple_all in future versions. To silence the warning, set filter_func explicitly.

max_relative_violations indicates what fraction of rows of the given table may have values not included in the reference set of unique values. Please note that UniquesSubset and UniquesSuperset are not symmetrical in this regard.

By default, the number of occurrences affects the computed fraction of violations. To disable this weighting, set compare_distinct=True. This argument does not have an effect on the test results for other Uniques constraints, or if max_relative_violations is 0.

See Uniques for further details on map_func, reduce_func, and output_processors.

Source code in src/datajudge/requirements.py

def add_uniques_subset_constraint(
    self,
    columns: list[str],
    uniques: Collection[_T],
    max_relative_violations: float = 0,
    filter_func: Callable[[list[_T]], list[_T]] | None = None,
    compare_distinct: bool = False,
    map_func: Callable[[_T], _T] | None = None,
    reduce_func: Callable[[Collection], Collection] | None = None,
    condition: Condition | None = None,
    name: str | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    cache_size=None,
):
    """Check if the data's unique values are contained in a given set of values.

    The ``UniquesSubset`` constraint asserts if the values contained in a column of
    a ``DataSource`` are part of a reference set of expected values, specified via
    ``uniques``.

    Null values in the columns ``columns`` are ignored. To assert the non-existence of them use
    the [`add_null_absence_constraint`][datajudge.requirements.WithinRequirement.add_null_absence_constraint] helper method
    for ``WithinRequirement``.
    By default, the null filtering does not trigger if multiple columns are fetched at once.
    It can be configured in more detail by supplying a custom ``filter_func`` function.
    Some exemplary implementations are available as [`filternull_element`][datajudge.utils.filternull_element],
    [`filternull_never`][datajudge.utils.filternull_never], [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all],
    [`filternull_element_or_tuple_any`][datajudge.utils.filternull_element_or_tuple_any].
    Passing ``None`` as the argument is equivalent to [`filternull_element`][datajudge.utils.filternull_element] but triggers a warning.
    The current default of [`filternull_element`][datajudge.utils.filternull_element]
    Cause (possibly often unintended) changes in behavior when the users adds a second column
    (filtering no longer can trigger at all).
    The default will be changed to [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all] in future versions.
    To silence the warning, set ``filter_func`` explicitly.


    ``max_relative_violations`` indicates what fraction of rows of the given table
    may have values not included in the reference set of unique values. Please note
    that ``UniquesSubset`` and ``UniquesSuperset`` are not symmetrical in this regard.

    By default, the number of occurrences affects the computed fraction of violations.
    To disable this weighting, set `compare_distinct=True`.
    This argument does not have an effect on the test results for other `Uniques` constraints,
    or if `max_relative_violations` is 0.

    See ``Uniques`` for further details on ``map_func``, ``reduce_func``,
    and ``output_processors``.
    """
    ref = DataReference(self._data_source, columns, condition)
    self._constraints.append(
        uniques_constraints.UniquesSubset(
            ref,
            uniques=uniques,
            max_relative_violations=max_relative_violations,
            filter_func=filter_func,
            compare_distinct=compare_distinct,
            map_func=map_func,
            reduce_func=reduce_func,
            output_processors=output_processors,
            name=name,
            cache_size=cache_size,
        )
    )

add_uniques_superset_constraint

add_uniques_superset_constraint(columns: list[str], uniques: Collection[_T], max_relative_violations: float = 0, filter_func: Callable[[list[_T]], list[_T]] | None = None, map_func: Callable[[_T], _T] | None = None, reduce_func: Callable[[Collection], Collection] | None = None, condition: Condition | None = None, name: str | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, cache_size=None)

Check if unique values of columns are contained in the reference data.

The UniquesSuperset constraint asserts that reference set of expected values, specified via uniques, is contained in given columns of a DataSource.

Null values in the columns columns are ignored. To assert the non-existence of them use the add_null_absence_constraint helper method for WithinRequirement.

By default, the null filtering does not trigger if multiple columns are fetched at once. It can be configured in more detail by supplying a custom filter_func function. Some exemplary implementations are available as filternull_element, filternull_never, filternull_element_or_tuple_all, filternull_element_or_tuple_any. Passing None as the argument is equivalent to filternull_element but triggers a warning. The current default of filternull_element will cause (possibly often unintended) changes in behavior when the user adds a second column (filtering no longer can trigger at all). The default will be changed to filternull_element_or_tuple_all in future versions. To silence the warning, set filter_func explicitly.

max_relative_violations indicates what fraction of unique values of the given DataSource are not represented in the reference set of unique values. Please note that UniquesSubset and UniquesSuperset are not symmetrical in this regard.

One use of this constraint is to test for consistency in columns with expected categorical values.

See Uniques for further details on map_func, reduce_func, and output_processors.

Source code in src/datajudge/requirements.py

def add_uniques_superset_constraint(
    self,
    columns: list[str],
    uniques: Collection[_T],
    max_relative_violations: float = 0,
    filter_func: Callable[[list[_T]], list[_T]] | None = None,
    map_func: Callable[[_T], _T] | None = None,
    reduce_func: Callable[[Collection], Collection] | None = None,
    condition: Condition | None = None,
    name: str | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    cache_size=None,
):
    """Check if unique values of columns are contained in the reference data.

    The ``UniquesSuperset`` constraint asserts that reference set of expected values,
    specified via ``uniques``, is contained in given columns of a ``DataSource``.

    Null values in the columns ``columns`` are ignored. To assert the non-existence of them use
    the [`add_null_absence_constraint`][datajudge.requirements.WithinRequirement.add_null_absence_constraint] helper method
    for ``WithinRequirement``.

    By default, the null filtering does not trigger if multiple columns are fetched at once.
    It can be configured in more detail by supplying a custom ``filter_func`` function.
    Some exemplary implementations are available as [`filternull_element`][datajudge.utils.filternull_element],
    [`filternull_never`][datajudge.utils.filternull_never], [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all],
    [`filternull_element_or_tuple_any`][datajudge.utils.filternull_element_or_tuple_any].
    Passing ``None`` as the argument is equivalent to [`filternull_element`][datajudge.utils.filternull_element] but triggers a warning.
    The current default of [`filternull_element`][datajudge.utils.filternull_element]
    will cause (possibly often unintended) changes in behavior when the user adds a second column
    (filtering no longer can trigger at all).
    The default will be changed to [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all] in future versions.
    To silence the warning, set ``filter_func`` explicitly.

    ``max_relative_violations`` indicates what fraction of unique values of the given
    ``DataSource`` are not represented in the reference set of unique values. Please
    note that ``UniquesSubset`` and ``UniquesSuperset`` are not symmetrical in this regard.

    One use of this constraint is to test for consistency in columns with expected
    categorical values.

    See ``Uniques`` for further details on ``map_func``, ``reduce_func``,
    and ``output_processors``.
    """
    ref = DataReference(self._data_source, columns, condition)
    self._constraints.append(
        uniques_constraints.UniquesSuperset(
            ref,
            uniques=uniques,
            max_relative_violations=max_relative_violations,
            filter_func=filter_func,
            map_func=map_func,
            reduce_func=reduce_func,
            output_processors=output_processors,
            name=name,
            cache_size=cache_size,
        )
    )

add_varchar_max_length_constraint

add_varchar_max_length_constraint(column: str, max_length: int, condition: Condition | None = None, name: str | None = None, cache_size=None)

Source code in src/datajudge/requirements.py

def add_varchar_max_length_constraint(
    self,
    column: str,
    max_length: int,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        varchar_constraints.VarCharMaxLength(
            ref,
            max_length=max_length,
            name=name,
            cache_size=cache_size,
        )
    )

add_varchar_min_length_constraint

add_varchar_min_length_constraint(column: str, min_length: int, condition: Condition | None = None, name: str | None = None, cache_size=None)

Source code in src/datajudge/requirements.py

def add_varchar_min_length_constraint(
    self,
    column: str,
    min_length: int,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        varchar_constraints.VarCharMinLength(
            ref,
            min_length=min_length,
            name=name,
            cache_size=cache_size,
        )
    )

add_varchar_regex_constraint

add_varchar_regex_constraint(column: str, regex: str, condition: Condition | None = None, name: str | None = None, allow_none: bool = False, relative_tolerance: float = 0.0, aggregated: bool = True, n_counterexamples: int = 5, cache_size=None)

Assesses whether the values in a column match a given regular expression pattern.

The option allow_none can be used in cases where the column is defined as nullable and contains null values.

How the tolerance factor is calculated can be controlled with the aggregated flag. When True, the tolerance is calculated using unique values. If not, the tolerance is calculated using all the instances of the data.

n_counterexamples defines how many counterexamples are displayed in an assertion text. If all counterexamples are meant to be shown, provide -1 as an argument.

When using this method, the regex matching will take place in memory. If instead, you would like the matching to take place in database which is typically faster and substantially more memory-saving, please consider using add_varchar_regex_constraint_db.

Source code in src/datajudge/requirements.py

def add_varchar_regex_constraint(
    self,
    column: str,
    regex: str,
    condition: Condition | None = None,
    name: str | None = None,
    allow_none: bool = False,
    relative_tolerance: float = 0.0,
    aggregated: bool = True,
    n_counterexamples: int = 5,
    cache_size=None,
):
    """Assesses whether the values in a column match a given regular expression pattern.

    The option ``allow_none`` can be used in cases where the column is defined as
    nullable and contains null values.

    How the tolerance factor is calculated can be controlled with the ``aggregated``
    flag. When ``True``, the tolerance is calculated using unique values. If not, the
    tolerance is calculated using all the instances of the data.

    ``n_counterexamples`` defines how many counterexamples are displayed in an
    assertion text. If all counterexamples are meant to be shown, provide ``-1`` as
    an argument.

    When using this method, the regex matching will take place in memory. If instead,
    you would like the matching to take place in database which is typically faster and
    substantially more memory-saving, please consider using
    ``add_varchar_regex_constraint_db``.
    """
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        varchar_constraints.VarCharRegex(
            ref,
            regex,
            allow_none=allow_none,
            relative_tolerance=relative_tolerance,
            aggregated=aggregated,
            n_counterexamples=n_counterexamples,
            name=name,
            cache_size=cache_size,
        )
    )

add_varchar_regex_constraint_db

add_varchar_regex_constraint_db(column: str, regex: str, condition: Condition | None = None, name: str | None = None, relative_tolerance: float = 0.0, aggregated: bool = True, n_counterexamples: int = 5, cache_size=None)

Assesses whether the values in a column match a given regular expression pattern.

How the tolerance factor is calculated can be controlled with the aggregated flag. When True, the tolerance is calculated using unique values. If not, the tolerance is calculated using all the instances of the data.

n_counterexamples defines how many counterexamples are displayed in an assertion text. If all counterexamples are meant to be shown, provide -1 as an argument.

When using this method, the regex matching will take place in database, which is only supported for Postgres, Sqllite and Snowflake. Note that for this feature is only for Snowflake when using sqlalchemy-snowflake >= 1.4.0. As an alternative, add_varchar_regex_constraint performs the regex matching in memory. This is typically slower and more expensive in terms of memory but available on all supported database mamangement systems.

Source code in src/datajudge/requirements.py

def add_varchar_regex_constraint_db(
    self,
    column: str,
    regex: str,
    condition: Condition | None = None,
    name: str | None = None,
    relative_tolerance: float = 0.0,
    aggregated: bool = True,
    n_counterexamples: int = 5,
    cache_size=None,
):
    """Assesses whether the values in a column match a given regular expression pattern.

    How the tolerance factor is calculated can be controlled with the ``aggregated``
    flag. When ``True``, the tolerance is calculated using unique values. If not, the
    tolerance is calculated using all the instances of the data.

    ``n_counterexamples`` defines how many counterexamples are displayed in an
    assertion text. If all counterexamples are meant to be shown, provide ``-1`` as
    an argument.

    When using this method, the regex matching will take place in database, which is
    only supported for Postgres, Sqllite and Snowflake. Note that for this
    feature is only for Snowflake when using sqlalchemy-snowflake >= 1.4.0. As an
    alternative, ``add_varchar_regex_constraint`` performs the regex matching in memory.
    This is typically slower and more expensive in terms of memory but available
    on all supported database mamangement systems.
    """
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        varchar_constraints.VarCharRegexDb(
            ref,
            regex=regex,
            relative_tolerance=relative_tolerance,
            aggregated=aggregated,
            n_counterexamples=n_counterexamples,
            name=name,
            cache_size=cache_size,
        )
    )

from_expression classmethod

from_expression(expression: FromClause, name: str)

Create a WithinRequirement based on a sqlalchemy expression.

Any sqlalchemy object implementing the alias method can be passed as an argument for the expression parameter. This could, e.g. be an sqlalchemy.Table object or the result of a sqlalchemy.select call.

The name will be used to represent this expression in error messages.

Source code in src/datajudge/requirements.py

@classmethod
def from_expression(cls, expression: sa.sql.selectable.FromClause, name: str):
    """Create a ``WithinRequirement`` based on a sqlalchemy expression.

    Any sqlalchemy object implementing the ``alias`` method can be passed as an
    argument for the ``expression`` parameter. This could, e.g. be an
    ``sqlalchemy.Table`` object or the result of a ``sqlalchemy.select`` call.

    The ``name`` will be used to represent this expression in error messages.
    """
    return cls(data_source=ExpressionDataSource(expression, name))

from_raw_query classmethod

from_raw_query(query: str, name: str, columns: list[str] | None = None)

Create a WithinRequirement based on a raw query string.

The query parameter can be passed any query string returning rows, e.g. "SELECT * FROM myschema.mytable LIMIT 1337" or "SELECT id, name FROM table1 UNION SELECT id, name FROM table2".

The name will be used to represent this query in error messages.

If constraints rely on specific columns, these should be provided here via columns, e.g. ["id", "name"].

Source code in src/datajudge/requirements.py

@classmethod
def from_raw_query(cls, query: str, name: str, columns: list[str] | None = None):
    """Create a ``WithinRequirement`` based on a raw query string.

    The ``query`` parameter can be passed any query string returning rows, e.g.
    ``"SELECT * FROM myschema.mytable LIMIT 1337"`` or
    ``"SELECT id, name FROM table1 UNION SELECT id, name FROM table2"``.

    The ``name`` will be used to represent this query in error messages.

    If constraints rely on specific columns, these should be provided here via
    ``columns``, e.g. ``["id", "name"]``.
    """
    return cls(data_source=RawQueryDataSource(query, name, columns=columns))

from_table classmethod

from_table(db_name: str, schema_name: str, table_name: str)

Create a WithinRequirement based on a table.

Source code in src/datajudge/requirements.py

@classmethod
def from_table(cls, db_name: str, schema_name: str, table_name: str):
    """Create a `WithinRequirement` based on a table."""
    return cls(
        data_source=TableDataSource(
            db_name=db_name, schema_name=schema_name, table_name=table_name
        )
    )

condition

Condition dataclass

Condition(raw_string: str | None = None, conditions: Sequence[Condition] | None = None, reduction_operator: str | None = None)

Condition allows for further narrowing down of a DataSource in a Constraint.

A Condition can be thought of as a filter, the content of a sql 'where' clause or a condition as known from probability theory.

While a DataSource is expressed more generally, one might be interested in testing properties of a specific part of said DataSource in light of a particular constraint. Hence using Condition allows for the reusage of a DataSource, in lieu of creating a new custom DataSource with the Condition implicitly built in.

A Condition can either be 'atomic', i.e. not further reducible to sub-conditions or 'composite', i.e. combining multiple subconditions. In the former case, it can be instantiated with help of the raw_string parameter, e.g. "col1 > 0". In the latter case, it can be instantiated with help of the conditions and reduction_operator parameters. reduction_operator allows for two values: "and" (logical conjunction) and "or" (logical disjunction). Note that composition of Condition supports arbitrary degrees of nesting.

conditions class-attribute instance-attribute

conditions: Sequence[Condition] | None = None

raw_string class-attribute instance-attribute

raw_string: str | None = None

reduction_operator class-attribute instance-attribute

reduction_operator: str | None = None

constraints

base

Constraint

Constraint(ref: DataReference, *, ref2: DataReference | None = None, ref_value: Any = None, name: str | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, cache_size=None)

Bases: ABC

Express a DataReference constraint against either another DataReference or a reference value.

Constraints against other DataReferences are typically referred to as 'between' constraints. Please use the the ref2 argument to instantiate such a constraint. Constraints against a fixed reference value are typically referred to as 'within' constraints. Please use the ref_value argument to instantiate such a constraint.

A constraint typically relies on the comparison of factual and target values. The former represent the key quantity of interest as seen in the database, the latter the key quantity of interest as expected a priori. Such a comparison is meant to be carried out in the test method.

In order to obtain such values, the retrieve method defines a mapping from DataReference, be it the DataReference of primary interest, ref, or a baseline DataReference, ref2, to value. If ref_value is already provided, usually no further mapping needs to be taken care of.

By default, retrieved arguments are cached indefinitely @lru_cache(maxsize=None). This can be controlled by setting the cache_size argument to a different value. 0 disables caching.

Source code in src/datajudge/constraints/base.py

def __init__(
    self,
    ref: DataReference,
    *,
    ref2: DataReference | None = None,
    ref_value: Any = None,
    name: str | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    cache_size=None,
):
    self._check_if_valid_between_or_within(ref2, ref_value)
    self._ref = ref
    self._ref2 = ref2
    self._ref_value = ref_value
    self.name = name
    self._factual_selections: _OptionalSelections = None
    self._target_selections: _OptionalSelections = None
    self._factual_queries: list[str] | None = None
    self._target_queries: list[str] | None = None

    if (output_processors is not None) and (
        not isinstance(output_processors, list)
    ):
        output_processors = [output_processors]
    self._output_processors = output_processors

    self._cache_size = cache_size
    self._setup_caching()

name instance-attribute

name = name

get_description

get_description() -> str

Source code in src/datajudge/constraints/base.py

def get_description(self) -> str:
    if self.name is not None:
        return self.name
    if self._ref2 is None:
        data_source_string = str(self._ref.data_source)
    else:
        data_source1_string = str(self._ref.data_source)
        data_source2_string = str(self._ref2.data_source)

        data_source1_substring, data_source2_substring = _uncommon_substrings(
            data_source1_string, data_source2_string
        )
        data_source_string = f"{data_source1_substring} | {data_source2_substring}"
    return self.__class__.__name__ + "::" + data_source_string

test

test(engine: Engine) -> TestResult

Source code in src/datajudge/constraints/base.py

def test(self, engine: sa.engine.Engine) -> TestResult:
    value_factual = self._get_factual_value(engine)
    value_target = self._get_target_value(engine)
    is_success, assertion_message = self._compare(value_factual, value_target)
    if is_success:
        return TestResult.success()

    factual_queries = None
    if self._factual_selections:
        factual_queries = [
            str(
                factual_selection.compile(
                    engine, compile_kwargs={"literal_binds": True}
                )
            )
            for factual_selection in self._factual_selections
        ]
    target_queries = None
    if self._target_selections:
        target_queries = [
            str(
                target_selection.compile(
                    engine, compile_kwargs={"literal_binds": True}
                )
            )
            for target_selection in self._target_selections
        ]
    return TestResult.failure(
        assertion_message,
        self.get_description(),
        factual_queries,
        target_queries,
    )

TestResult dataclass

TestResult(outcome: bool, _failure_message: str | None = None, _constraint_description: str | None = None, _factual_queries: str | None = None, _target_queries: str | None = None)

The result of the execution of a Constraint.

constraint_description property

constraint_description: str | None

failure_message property

failure_message: str | None

logging_message property

logging_message

outcome instance-attribute

outcome: bool

failure classmethod

failure(*args, **kwargs)

Source code in src/datajudge/constraints/base.py

@classmethod
def failure(cls, *args, **kwargs):
    return cls(False, *args, **kwargs)

formatted_constraint_description

formatted_constraint_description(formatter: Formatter) -> str | None

Source code in src/datajudge/constraints/base.py

def formatted_constraint_description(self, formatter: Formatter) -> str | None:
    return (
        formatter.fmt_str(self._constraint_description)
        if self._constraint_description
        else None
    )

formatted_failure_message

formatted_failure_message(formatter: Formatter) -> str | None

Source code in src/datajudge/constraints/base.py

def formatted_failure_message(self, formatter: Formatter) -> str | None:
    return (
        formatter.fmt_str(self._failure_message) if self._failure_message else None
    )

success classmethod

success()

Source code in src/datajudge/constraints/base.py

@classmethod
def success(cls):
    return cls(True)

column

Column

Column(ref: DataReference, *, ref2: DataReference | None = None, ref_value: Any = None, name: str | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, cache_size=None)

Bases: Constraint, ABC

Source code in src/datajudge/constraints/base.py

def __init__(
    self,
    ref: DataReference,
    *,
    ref2: DataReference | None = None,
    ref_value: Any = None,
    name: str | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    cache_size=None,
):
    self._check_if_valid_between_or_within(ref2, ref_value)
    self._ref = ref
    self._ref2 = ref2
    self._ref_value = ref_value
    self.name = name
    self._factual_selections: _OptionalSelections = None
    self._target_selections: _OptionalSelections = None
    self._factual_queries: list[str] | None = None
    self._target_queries: list[str] | None = None

    if (output_processors is not None) and (
        not isinstance(output_processors, list)
    ):
        output_processors = [output_processors]
    self._output_processors = output_processors

    self._cache_size = cache_size
    self._setup_caching()

ColumnExistence

ColumnExistence(ref: DataReference, columns: list[str], name: str | None = None, cache_size=None)

Bases: Column

Source code in src/datajudge/constraints/column.py

def __init__(
    self,
    ref: DataReference,
    columns: list[str],
    name: str | None = None,
    cache_size=None,
):
    super().__init__(ref, ref_value=columns, name=name, cache_size=cache_size)

ColumnSubset

ColumnSubset(ref: DataReference, *, ref2: DataReference | None = None, ref_value: Any = None, name: str | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, cache_size=None)

Bases: Column

Source code in src/datajudge/constraints/base.py

def __init__(
    self,
    ref: DataReference,
    *,
    ref2: DataReference | None = None,
    ref_value: Any = None,
    name: str | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    cache_size=None,
):
    self._check_if_valid_between_or_within(ref2, ref_value)
    self._ref = ref
    self._ref2 = ref2
    self._ref_value = ref_value
    self.name = name
    self._factual_selections: _OptionalSelections = None
    self._target_selections: _OptionalSelections = None
    self._factual_queries: list[str] | None = None
    self._target_queries: list[str] | None = None

    if (output_processors is not None) and (
        not isinstance(output_processors, list)
    ):
        output_processors = [output_processors]
    self._output_processors = output_processors

    self._cache_size = cache_size
    self._setup_caching()

ColumnSuperset

ColumnSuperset(ref: DataReference, *, ref2: DataReference | None = None, ref_value: Any = None, name: str | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, cache_size=None)

Bases: Column

Source code in src/datajudge/constraints/base.py

def __init__(
    self,
    ref: DataReference,
    *,
    ref2: DataReference | None = None,
    ref_value: Any = None,
    name: str | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    cache_size=None,
):
    self._check_if_valid_between_or_within(ref2, ref_value)
    self._ref = ref
    self._ref2 = ref2
    self._ref_value = ref_value
    self.name = name
    self._factual_selections: _OptionalSelections = None
    self._target_selections: _OptionalSelections = None
    self._factual_queries: list[str] | None = None
    self._target_queries: list[str] | None = None

    if (output_processors is not None) and (
        not isinstance(output_processors, list)
    ):
        output_processors = [output_processors]
    self._output_processors = output_processors

    self._cache_size = cache_size
    self._setup_caching()

ColumnType

ColumnType(ref: DataReference, *, ref2: DataReference | None = None, column_type: str | TypeEngine | None = None, name: str | None = None, cache_size=None)

Bases: Constraint

A class used to represent a ColumnType constraint.

This class enables flexible specification of column types either in string format or using SQLAlchemy's type hierarchy. It checks whether a column's type matches the specified type, allowing for checks against backend-specific types, SQLAlchemy's generic types, or string representations of backend-specific types.

When using SQLAlchemy's generic types, the comparison is done using isinstance, which means that the actual type can also be a subclass of the target type. For more information, see https://docs.sqlalchemy.org/en/20/core/type_basics.html

Source code in src/datajudge/constraints/column.py

def __init__(
    self,
    ref: DataReference,
    *,
    ref2: DataReference | None = None,
    column_type: str | sa.types.TypeEngine | None = None,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=column_type,
        name=name,
        cache_size=cache_size,
    )

date

Date module-attribute

Date = str | date | datetime

DateBetween

DateBetween(ref: DataReference, min_fraction: float, lower_bound: str, upper_bound: str, name: str | None = None, cache_size=None)

Bases: Constraint

Source code in src/datajudge/constraints/date.py

def __init__(
    self,
    ref: DataReference,
    min_fraction: float,
    lower_bound: str,
    upper_bound: str,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(ref, ref_value=min_fraction, name=name, cache_size=cache_size)
    self._lower_bound = lower_bound
    self._upper_bound = upper_bound

DateMax

DateMax(ref: DataReference, use_upper_bound_reference: bool, column_type: str, name: str | None = None, cache_size=None, *, ref2: DataReference | None = None, max_value: str | None = None)

Bases: Constraint

Source code in src/datajudge/constraints/date.py

def __init__(
    self,
    ref: DataReference,
    use_upper_bound_reference: bool,
    column_type: str,
    name: str | None = None,
    cache_size=None,
    *,
    ref2: DataReference | None = None,
    max_value: str | None = None,
):
    self._format = _get_format_from_column_type(column_type)
    self._use_upper_bound_reference = use_upper_bound_reference
    max_date: dt.date | None = None
    if max_value is not None:
        max_date = dt.datetime.strptime(max_value, _INPUT_DATE_FORMAT).date()
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=max_date,
        name=name,
        cache_size=cache_size,
    )

DateMin

DateMin(ref: DataReference, use_lower_bound_reference: bool, column_type: str, name: str | None = None, cache_size=None, *, ref2: DataReference | None = None, min_value: str | None = None)

Bases: Constraint

Source code in src/datajudge/constraints/date.py

def __init__(
    self,
    ref: DataReference,
    use_lower_bound_reference: bool,
    column_type: str,
    name: str | None = None,
    cache_size=None,
    *,
    ref2: DataReference | None = None,
    min_value: str | None = None,
):
    self._format = _get_format_from_column_type(column_type)
    self._use_lower_bound_reference = use_lower_bound_reference
    min_date: dt.date | None = None
    if min_value is not None:
        min_date = dt.datetime.strptime(min_value, _INPUT_DATE_FORMAT).date()
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=min_date,
        name=name,
        cache_size=cache_size,
    )

DateNoGap

DateNoGap(ref: DataReference, key_columns: list[str] | None, start_columns: list[str], end_columns: list[str], max_relative_n_violations: float, legitimate_gap_size: float, name: str | None = None, cache_size=None)

Bases: NoGapConstraint

Source code in src/datajudge/constraints/interval.py

def __init__(
    self,
    ref: DataReference,
    key_columns: list[str] | None,
    start_columns: list[str],
    end_columns: list[str],
    max_relative_n_violations: float,
    legitimate_gap_size: float,
    name: str | None = None,
    cache_size=None,
):
    self._legitimate_gap_size = legitimate_gap_size
    super().__init__(
        ref,
        key_columns,
        start_columns,
        end_columns,
        max_relative_n_violations,
        name=name,
        cache_size=cache_size,
    )

select

select(engine: Engine, ref: DataReference)

Source code in src/datajudge/constraints/date.py

def select(self, engine: sa.engine.Engine, ref: DataReference):
    sample_selection, n_violations_selection = db_access.get_date_gaps(
        engine,
        ref,
        self._key_columns,
        self._start_columns[0],
        self._end_columns[0],
        self._legitimate_gap_size,
    )
    # TODO: Once get_unique_count also only returns a selection without
    # executing it, one would want to list this selection here as well.
    return sample_selection, n_violations_selection

DateNoOverlap

DateNoOverlap(ref: DataReference, key_columns: list[str] | None, start_columns: list[str], end_columns: list[str], max_relative_n_violations: float, end_included: bool, name: str | None = None, cache_size=None)

Bases: NoOverlapConstraint

Source code in src/datajudge/constraints/interval.py

def __init__(
    self,
    ref: DataReference,
    key_columns: list[str] | None,
    start_columns: list[str],
    end_columns: list[str],
    max_relative_n_violations: float,
    end_included: bool,
    name: str | None = None,
    cache_size=None,
):
    self._end_included = end_included
    super().__init__(
        ref,
        key_columns,
        start_columns,
        end_columns,
        max_relative_n_violations,
        name=name,
        cache_size=cache_size,
    )

DateNoOverlap2d

DateNoOverlap2d(ref: DataReference, key_columns: list[str] | None, start_columns: list[str], end_columns: list[str], max_relative_n_violations: float, end_included: bool, name: str | None = None, cache_size=None)

Bases: NoOverlapConstraint

Source code in src/datajudge/constraints/interval.py

def __init__(
    self,
    ref: DataReference,
    key_columns: list[str] | None,
    start_columns: list[str],
    end_columns: list[str],
    max_relative_n_violations: float,
    end_included: bool,
    name: str | None = None,
    cache_size=None,
):
    self._end_included = end_included
    super().__init__(
        ref,
        key_columns,
        start_columns,
        end_columns,
        max_relative_n_violations,
        name=name,
        cache_size=cache_size,
    )

groupby

AggregateNumericRangeEquality

AggregateNumericRangeEquality(ref: DataReference, aggregation_column: str, start_value: int = 0, name: str | None = None, cache_size=None, *, tolerance: float = 0, ref2: DataReference | None = None)

Bases: Constraint

Source code in src/datajudge/constraints/groupby.py

def __init__(
    self,
    ref: DataReference,
    aggregation_column: str,
    start_value: int = 0,
    name: str | None = None,
    cache_size=None,
    *,
    tolerance: float = 0,
    ref2: DataReference | None = None,
):
    super().__init__(ref, ref2=ref2, ref_value=object(), name=name)
    self._aggregation_column = aggregation_column
    self._tolerance = tolerance
    self._start_value = start_value
    self._selection = None

interval

IntervalConstraint

IntervalConstraint(ref: DataReference, key_columns: list[str] | None, start_columns: list[str], end_columns: list[str], max_relative_n_violations: float, name: str | None = None, cache_size=None)

Bases: Constraint

Source code in src/datajudge/constraints/interval.py

def __init__(
    self,
    ref: DataReference,
    key_columns: list[str] | None,
    start_columns: list[str],
    end_columns: list[str],
    max_relative_n_violations: float,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(ref, ref_value=object(), name=name)
    self._key_columns = key_columns
    self._start_columns = start_columns
    self._end_columns = end_columns
    self._max_relative_n_violations = max_relative_n_violations
    self._validate_dimensions()

select abstractmethod

select(engine: Engine, ref: DataReference)

Source code in src/datajudge/constraints/interval.py

@abc.abstractmethod
def select(self, engine: sa.engine.Engine, ref: DataReference):
    pass

NoGapConstraint

NoGapConstraint(ref: DataReference, key_columns: list[str] | None, start_columns: list[str], end_columns: list[str], max_relative_n_violations: float, legitimate_gap_size: float, name: str | None = None, cache_size=None)

Bases: IntervalConstraint

Source code in src/datajudge/constraints/interval.py

def __init__(
    self,
    ref: DataReference,
    key_columns: list[str] | None,
    start_columns: list[str],
    end_columns: list[str],
    max_relative_n_violations: float,
    legitimate_gap_size: float,
    name: str | None = None,
    cache_size=None,
):
    self._legitimate_gap_size = legitimate_gap_size
    super().__init__(
        ref,
        key_columns,
        start_columns,
        end_columns,
        max_relative_n_violations,
        name=name,
        cache_size=cache_size,
    )

select abstractmethod

select(engine: Engine, ref: DataReference)

Source code in src/datajudge/constraints/interval.py

@abc.abstractmethod
def select(self, engine: sa.engine.Engine, ref: DataReference):
    pass

NoOverlapConstraint

NoOverlapConstraint(ref: DataReference, key_columns: list[str] | None, start_columns: list[str], end_columns: list[str], max_relative_n_violations: float, end_included: bool, name: str | None = None, cache_size=None)

Bases: IntervalConstraint

Source code in src/datajudge/constraints/interval.py

def __init__(
    self,
    ref: DataReference,
    key_columns: list[str] | None,
    start_columns: list[str],
    end_columns: list[str],
    max_relative_n_violations: float,
    end_included: bool,
    name: str | None = None,
    cache_size=None,
):
    self._end_included = end_included
    super().__init__(
        ref,
        key_columns,
        start_columns,
        end_columns,
        max_relative_n_violations,
        name=name,
        cache_size=cache_size,
    )

select

select(engine: Engine, ref: DataReference)

Source code in src/datajudge/constraints/interval.py

def select(self, engine: sa.engine.Engine, ref: DataReference):
    sample_selection, n_violations_selection = db_access.get_interval_overlaps_nd(
        engine,
        ref,
        self._key_columns,
        start_columns=self._start_columns,
        end_columns=self._end_columns,
        end_included=self._end_included,
    )
    # TODO: Once get_unique_count also only returns a selection without
    # executing it, one would want to list this selection here as well.
    return sample_selection, n_violations_selection

miscs

FunctionalDependency

FunctionalDependency(ref: DataReference, key_columns: list[str], **kwargs)

Bases: Constraint

Source code in src/datajudge/constraints/miscs.py

def __init__(self, ref: DataReference, key_columns: list[str], **kwargs):
    super().__init__(ref, ref_value=object(), **kwargs)
    self.key_columns = key_columns

key_columns instance-attribute

key_columns = key_columns

test

test(engine: Engine) -> TestResult

Source code in src/datajudge/constraints/miscs.py

def test(self, engine: sa.engine.Engine) -> TestResult:
    violations, _ = db_access.get_functional_dependency_violations(
        engine, self._ref, self.key_columns
    )
    if not violations:
        return TestResult.success()

    assertion_text = (
        f"{self._ref} has violations of functional dependence (in total {len(violations)} rows):\n"
        + "\n".join(
            [
                f"{violation}"
                for violation in self._apply_output_formatting(
                    [tuple(elem) for elem in violations]
                )
            ]
        )
    )
    return TestResult.failure(assertion_text)

MaxNullFraction

MaxNullFraction(ref, *, ref2: DataReference | None = None, max_null_fraction: float | None = None, max_relative_deviation: float = 0, name: str | None = None, cache_size=None)

Bases: Constraint

Source code in src/datajudge/constraints/miscs.py

def __init__(
    self,
    ref,
    *,
    ref2: DataReference | None = None,
    max_null_fraction: float | None = None,
    max_relative_deviation: float = 0,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=max_null_fraction,
        name=name,
        cache_size=cache_size,
    )
    if max_null_fraction is not None and not (0 <= max_null_fraction <= 1):
        raise ValueError(
            f"max_null_fraction was expected to lie within [0, 1] but is "
            f"{max_null_fraction}."
        )
    if max_relative_deviation < 0:
        raise ValueError(
            f"{max_relative_deviation} is negative even though it needs to be positive."
        )
    self.max_relative_deviation = max_relative_deviation

max_relative_deviation instance-attribute

max_relative_deviation = max_relative_deviation

PrimaryKeyDefinition

PrimaryKeyDefinition(ref, primary_keys: list[str], name: str | None = None, cache_size=None)

Bases: Constraint

Source code in src/datajudge/constraints/miscs.py

def __init__(
    self,
    ref,
    primary_keys: list[str],
    name: str | None = None,
    cache_size=None,
):
    super().__init__(ref, ref_value=set(primary_keys), name=name)

Uniqueness

Uniqueness(ref: DataReference, max_duplicate_fraction: float = 0, max_absolute_n_duplicates: int = 0, infer_pk_columns: bool = False, name: str | None = None, cache_size=None)

Bases: Constraint

Source code in src/datajudge/constraints/miscs.py

def __init__(
    self,
    ref: DataReference,
    max_duplicate_fraction: float = 0,
    max_absolute_n_duplicates: int = 0,
    infer_pk_columns: bool = False,
    name: str | None = None,
    cache_size=None,
):
    if max_duplicate_fraction != 0 and max_absolute_n_duplicates != 0:
        raise ValueError(
            """Uniqueness constraint was attempted to be constructed
            with both a relative and an absolute tolerance. Only use one
            of both at a time."""
        )
    if max_duplicate_fraction != 0:
        ref_value = ("relative", max_duplicate_fraction)
    elif max_absolute_n_duplicates != 0:
        ref_value = ("absolute", max_absolute_n_duplicates)
    else:
        ref_value = ("relative", 0)

    self.infer_pk_columns = infer_pk_columns
    super().__init__(ref, ref_value=ref_value, name=name, cache_size=cache_size)

infer_pk_columns instance-attribute

infer_pk_columns = infer_pk_columns

test

test(engine: Engine) -> TestResult

Source code in src/datajudge/constraints/miscs.py

def test(self, engine: sa.engine.Engine) -> TestResult:
    if self.infer_pk_columns and db_access.is_bigquery(engine):
        raise NotImplementedError("No primary key concept in BigQuery")

    # only check for primary keys when actually defined
    # otherwise default back to searching the whole table
    if self.infer_pk_columns and (
        pk_columns := db_access.get_primary_keys(engine, self._ref)[0]
    ):
        self._ref.columns = pk_columns
        if not pk_columns:  # there were no primary keys found
            warnings.warn(
                f"""No primary keys found in {self._ref}.
                Uniqueness will be tested for all columns."""
            )

    unique_count, unique_selections = db_access.get_unique_count(engine, self._ref)
    row_count, row_selections = db_access.get_row_count(engine, self._ref)
    self.factual_selections = row_selections
    self.target_selections = unique_selections
    if row_count == 0:
        return TestResult(True, "No occurrences.")
    tolerance_kind, tolerance_value = self._ref_value  # type: ignore
    if tolerance_kind == "relative":
        result = unique_count >= row_count * (1 - tolerance_value)
    elif tolerance_kind == "absolute":
        result = unique_count >= row_count - tolerance_value
    else:
        raise ValueError(
            "Given tolerance is neither relative nor absolute: {tolerance_kind}."
        )
    if result:
        return TestResult.success()
    sample, _ = db_access.get_duplicate_sample(engine, self._ref)
    sample_string = _format_sample(sample, self._ref)
    assertion_text = (
        f"{self._ref} has {row_count} rows > {unique_count} "
        f"uniques. This surpasses the max_duplicate_fraction of "
        f"{self._ref_value}. An example tuple breaking the "
        f"uniqueness condition is: {sample_string}."
    )
    return TestResult.failure(assertion_text)

nrows

NRows

NRows(ref: DataReference, *, ref2: DataReference | None = None, n_rows: int | None = None, name: str | None = None, cache_size=None)

Bases: Constraint, ABC

Source code in src/datajudge/constraints/nrows.py

def __init__(
    self,
    ref: DataReference,
    *,
    ref2: DataReference | None = None,
    n_rows: int | None = None,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=n_rows,
        name=name,
        cache_size=cache_size,
    )

NRowsEquality

NRowsEquality(ref: DataReference, *, ref2: DataReference | None = None, n_rows: int | None = None, name: str | None = None, cache_size=None)

Bases: NRows

Source code in src/datajudge/constraints/nrows.py

def __init__(
    self,
    ref: DataReference,
    *,
    ref2: DataReference | None = None,
    n_rows: int | None = None,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=n_rows,
        name=name,
        cache_size=cache_size,
    )

NRowsMax

NRowsMax(ref: DataReference, *, ref2: DataReference | None = None, n_rows: int | None = None, name: str | None = None, cache_size=None)

Bases: NRows

Source code in src/datajudge/constraints/nrows.py

def __init__(
    self,
    ref: DataReference,
    *,
    ref2: DataReference | None = None,
    n_rows: int | None = None,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=n_rows,
        name=name,
        cache_size=cache_size,
    )

NRowsMaxGain

NRowsMaxGain(ref: DataReference, ref2: DataReference, max_relative_gain_getter: _ToleranceGetter, name: str | None = None, cache_size=None)

Bases: NRows

Source code in src/datajudge/constraints/nrows.py

def __init__(
    self,
    ref: DataReference,
    ref2: DataReference,
    max_relative_gain_getter: _ToleranceGetter,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(ref, ref2=ref2, name=name, cache_size=cache_size)
    self.max_relative_gain_getter = max_relative_gain_getter

max_relative_gain_getter instance-attribute

max_relative_gain_getter = max_relative_gain_getter

test

test(engine: Engine) -> TestResult

Source code in src/datajudge/constraints/nrows.py

def test(self, engine: sa.engine.Engine) -> TestResult:
    self.max_relative_gain = self.max_relative_gain_getter(engine)
    return super().test(engine)

NRowsMaxLoss

NRowsMaxLoss(ref: DataReference, ref2: DataReference, max_relative_loss_getter: _ToleranceGetter, name: str | None = None, cache_size=None)

Bases: NRows

Source code in src/datajudge/constraints/nrows.py

def __init__(
    self,
    ref: DataReference,
    ref2: DataReference,
    max_relative_loss_getter: _ToleranceGetter,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(ref, ref2=ref2, name=name, cache_size=cache_size)
    self.max_relative_loss_getter = max_relative_loss_getter

max_relative_loss_getter instance-attribute

max_relative_loss_getter = max_relative_loss_getter

test

test(engine: Engine) -> TestResult

Source code in src/datajudge/constraints/nrows.py

def test(self, engine: sa.engine.Engine) -> TestResult:
    self.max_relative_loss = self.max_relative_loss_getter(engine)
    return super().test(engine)

NRowsMin

NRowsMin(ref: DataReference, *, ref2: DataReference | None = None, n_rows: int | None = None, name: str | None = None, cache_size=None)

Bases: NRows

Source code in src/datajudge/constraints/nrows.py

def __init__(
    self,
    ref: DataReference,
    *,
    ref2: DataReference | None = None,
    n_rows: int | None = None,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=n_rows,
        name=name,
        cache_size=cache_size,
    )

NRowsMinGain

NRowsMinGain(ref: DataReference, ref2: DataReference, min_relative_gain_getter: _ToleranceGetter, name: str | None = None, cache_size=None)

Bases: NRows

Source code in src/datajudge/constraints/nrows.py

def __init__(
    self,
    ref: DataReference,
    ref2: DataReference,
    min_relative_gain_getter: _ToleranceGetter,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(ref, ref2=ref2, name=name, cache_size=cache_size)
    self.min_relative_gain_getter = min_relative_gain_getter

min_relative_gain_getter instance-attribute

min_relative_gain_getter = min_relative_gain_getter

test

test(engine: Engine) -> TestResult

Source code in src/datajudge/constraints/nrows.py

def test(self, engine: sa.engine.Engine) -> TestResult:
    self.min_relative_gain = self.min_relative_gain_getter(engine)
    return super().test(engine)

numeric

NumericBetween

NumericBetween(ref: DataReference, min_fraction: float, lower_bound: float, upper_bound: float, name: str | None = None, cache_size=None)

Bases: Constraint

Source code in src/datajudge/constraints/numeric.py

def __init__(
    self,
    ref: DataReference,
    min_fraction: float,
    lower_bound: float,
    upper_bound: float,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(ref, ref_value=min_fraction, name=name, cache_size=cache_size)
    self._lower_bound = lower_bound
    self._upper_bound = upper_bound

NumericMax

NumericMax(ref: DataReference, name: str | None = None, cache_size=None, *, ref2: DataReference | None = None, max_value: float | None = None)

Bases: Constraint

Source code in src/datajudge/constraints/numeric.py

def __init__(
    self,
    ref: DataReference,
    name: str | None = None,
    cache_size=None,
    *,
    ref2: DataReference | None = None,
    max_value: float | None = None,
):
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=max_value,
        name=name,
        cache_size=cache_size,
    )

NumericMean

NumericMean(ref: DataReference, max_absolute_deviation: float, name: str | None = None, cache_size=None, *, ref2: DataReference | None = None, mean_value: float | None = None)

Bases: Constraint

Source code in src/datajudge/constraints/numeric.py

def __init__(
    self,
    ref: DataReference,
    max_absolute_deviation: float,
    name: str | None = None,
    cache_size=None,
    *,
    ref2: DataReference | None = None,
    mean_value: float | None = None,
):
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=mean_value,
        name=name,
        cache_size=cache_size,
    )
    self._max_absolute_deviation = max_absolute_deviation

test

test(engine: Engine) -> TestResult

Source code in src/datajudge/constraints/numeric.py

def test(self, engine: sa.engine.Engine) -> TestResult:
    mean_factual = self._get_factual_value(engine)
    mean_target = self._get_target_value(engine)
    if mean_factual is None or mean_target is None:
        return TestResult(
            mean_factual is None and mean_target is None,
            "Mean over empty set.",
        )
    deviation = abs(mean_factual - mean_target)
    assertion_text = (
        f"{self._ref} "
        f"has mean {mean_factual}, deviating more than "
        f"{self._max_absolute_deviation} from "
        f"{self._target_prefix} {mean_target}. "
        f"{self._condition_string}"
    )
    result = deviation <= self._max_absolute_deviation
    return TestResult(result, assertion_text)

NumericMin

NumericMin(ref: DataReference, name: str | None = None, cache_size=None, *, ref2: DataReference | None = None, min_value: float | None = None)

Bases: Constraint

Source code in src/datajudge/constraints/numeric.py

def __init__(
    self,
    ref: DataReference,
    name: str | None = None,
    cache_size=None,
    *,
    ref2: DataReference | None = None,
    min_value: float | None = None,
):
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=min_value,
        name=name,
        cache_size=cache_size,
    )

NumericNoGap

NumericNoGap(ref: DataReference, key_columns: list[str] | None, start_columns: list[str], end_columns: list[str], max_relative_n_violations: float, legitimate_gap_size: float, name: str | None = None, cache_size=None)

Bases: NoGapConstraint

Source code in src/datajudge/constraints/interval.py

def __init__(
    self,
    ref: DataReference,
    key_columns: list[str] | None,
    start_columns: list[str],
    end_columns: list[str],
    max_relative_n_violations: float,
    legitimate_gap_size: float,
    name: str | None = None,
    cache_size=None,
):
    self._legitimate_gap_size = legitimate_gap_size
    super().__init__(
        ref,
        key_columns,
        start_columns,
        end_columns,
        max_relative_n_violations,
        name=name,
        cache_size=cache_size,
    )

select

select(engine: Engine, ref: DataReference)

Source code in src/datajudge/constraints/numeric.py

def select(self, engine: sa.engine.Engine, ref: DataReference):
    sample_selection, n_violations_selection = db_access.get_numeric_gaps(
        engine,
        ref,
        self._key_columns,
        self._start_columns[0],
        self._end_columns[0],
        self._legitimate_gap_size,
    )
    # TODO: Once get_unique_count also only returns a selection without
    # executing it, one would want to list this selection here as well.
    return sample_selection, n_violations_selection

NumericNoOverlap

NumericNoOverlap(ref: DataReference, key_columns: list[str] | None, start_columns: list[str], end_columns: list[str], max_relative_n_violations: float, end_included: bool, name: str | None = None, cache_size=None)

Bases: NoOverlapConstraint

Source code in src/datajudge/constraints/interval.py

def __init__(
    self,
    ref: DataReference,
    key_columns: list[str] | None,
    start_columns: list[str],
    end_columns: list[str],
    max_relative_n_violations: float,
    end_included: bool,
    name: str | None = None,
    cache_size=None,
):
    self._end_included = end_included
    super().__init__(
        ref,
        key_columns,
        start_columns,
        end_columns,
        max_relative_n_violations,
        name=name,
        cache_size=cache_size,
    )

NumericPercentile

NumericPercentile(ref: DataReference, percentage: float, max_absolute_deviation: float | None = None, max_relative_deviation: float | None = None, name: str | None = None, cache_size=None, *, ref2: DataReference | None = None, expected_percentile: float | None = None)

Bases: Constraint

Source code in src/datajudge/constraints/numeric.py

def __init__(
    self,
    ref: DataReference,
    percentage: float,
    max_absolute_deviation: float | None = None,
    max_relative_deviation: float | None = None,
    name: str | None = None,
    cache_size=None,
    *,
    ref2: DataReference | None = None,
    expected_percentile: float | None = None,
):
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=expected_percentile,
        name=name,
        cache_size=cache_size,
    )
    if not (0 <= percentage <= 100):
        raise ValueError(
            f"Expected percentage to be a value between 0 and 100, got {percentage}."
        )
    self.percentage = percentage
    if max_absolute_deviation is None and max_relative_deviation is None:
        raise ValueError(
            "At least one of 'max_absolute_deviation' and 'max_relative_deviation' "
            "must be given."
        )
    if max_absolute_deviation is not None and max_absolute_deviation < 0:
        raise ValueError(
            f"max_absolute_deviation must be at least 0 but is {max_absolute_deviation}."
        )
    if max_relative_deviation is not None and max_relative_deviation < 0:
        raise ValueError(
            f"max_relative_deviation must be at least 0 but is {max_relative_deviation}."
        )
    self._max_absolute_deviation = max_absolute_deviation
    self._max_relative_deviation = max_relative_deviation

percentage instance-attribute

percentage = percentage

row

Row

Row(ref: DataReference, ref2: DataReference, max_missing_fraction_getter: _ToleranceGetter, name: str | None = None, cache_size=None)

Bases: Constraint, ABC

Source code in src/datajudge/constraints/row.py

def __init__(
    self,
    ref: DataReference,
    ref2: DataReference,
    max_missing_fraction_getter: _ToleranceGetter,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(ref, ref2=ref2, name=name, cache_size=cache_size)
    self._max_missing_fraction_getter = max_missing_fraction_getter

test

test(engine: Engine) -> TestResult

Source code in src/datajudge/constraints/row.py

def test(self, engine: sa.engine.Engine) -> TestResult:
    if self._ref is None or self._ref2 is None:
        raise ValueError()
    self.max_missing_fraction = self._max_missing_fraction_getter(engine)
    self._ref1_minus_ref2_sample, _ = db_access.get_row_difference_sample(
        engine, self._ref, self._ref2
    )
    self._ref2_minus_ref1_sample, _ = db_access.get_row_difference_sample(
        engine, self._ref2, self._ref
    )
    return super().test(engine)

RowEquality

RowEquality(ref: DataReference, ref2: DataReference, max_missing_fraction_getter: _ToleranceGetter, name: str | None = None, cache_size=None)

Bases: Row

Source code in src/datajudge/constraints/row.py

def __init__(
    self,
    ref: DataReference,
    ref2: DataReference,
    max_missing_fraction_getter: _ToleranceGetter,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(ref, ref2=ref2, name=name, cache_size=cache_size)
    self._max_missing_fraction_getter = max_missing_fraction_getter

RowMatchingEquality

RowMatchingEquality(ref: DataReference, ref2: DataReference, matching_columns1: list[str], matching_columns2: list[str], comparison_columns1: list[str], comparison_columns2: list[str], max_missing_fraction_getter: _ToleranceGetter, name: str | None = None, cache_size=None)

Bases: Row

Source code in src/datajudge/constraints/row.py

def __init__(
    self,
    ref: DataReference,
    ref2: DataReference,
    matching_columns1: list[str],
    matching_columns2: list[str],
    comparison_columns1: list[str],
    comparison_columns2: list[str],
    max_missing_fraction_getter: _ToleranceGetter,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(
        ref,
        ref2=ref2,
        max_missing_fraction_getter=max_missing_fraction_getter,
        name=name,
        cache_size=cache_size,
    )
    self._match_and_compare = _MatchAndCompare(
        matching_columns1,
        matching_columns2,
        comparison_columns1,
        comparison_columns2,
    )

test

test(engine: Engine) -> TestResult

Source code in src/datajudge/constraints/row.py

def test(self, engine: sa.engine.Engine) -> TestResult:
    if self._ref is None or self._ref2 is None:
        raise ValueError()
    missing_fraction, n_rows_match, selections = db_access.get_row_mismatch(
        engine, self._ref, self._ref2, self._match_and_compare
    )
    self.factual_selections = selections
    max_missing_fraction = self._max_missing_fraction_getter(engine)
    result = missing_fraction <= max_missing_fraction
    if result:
        return TestResult.success()
    assertion_message = (
        f"{missing_fraction} > "
        f"{max_missing_fraction} of the rows differ "
        f"on a match of {n_rows_match} rows between {self._ref} and "
        f"{self._ref2}. "
        f"{self._condition_string}"
        f"{self._match_and_compare} "
    )
    return TestResult.failure(assertion_message)

RowSubset

RowSubset(ref: DataReference, ref2: DataReference, max_missing_fraction_getter: _ToleranceGetter, name: str | None = None, cache_size=None)

Bases: Row

Source code in src/datajudge/constraints/row.py

def __init__(
    self,
    ref: DataReference,
    ref2: DataReference,
    max_missing_fraction_getter: _ToleranceGetter,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(ref, ref2=ref2, name=name, cache_size=cache_size)
    self._max_missing_fraction_getter = max_missing_fraction_getter

RowSuperset

RowSuperset(ref: DataReference, ref2: DataReference, max_missing_fraction_getter: _ToleranceGetter, name: str | None = None, cache_size=None)

Bases: Row

Source code in src/datajudge/constraints/row.py

def __init__(
    self,
    ref: DataReference,
    ref2: DataReference,
    max_missing_fraction_getter: _ToleranceGetter,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(ref, ref2=ref2, name=name, cache_size=cache_size)
    self._max_missing_fraction_getter = max_missing_fraction_getter

stats

KolmogorovSmirnov2Sample

KolmogorovSmirnov2Sample(ref: DataReference, ref2: DataReference, significance_level: float = 0.05, name: str | None = None, cache_size=None)

Bases: Constraint

Source code in src/datajudge/constraints/stats.py

def __init__(
    self,
    ref: DataReference,
    ref2: DataReference,
    significance_level: float = 0.05,
    name: str | None = None,
    cache_size=None,
):
    self._significance_level = significance_level
    super().__init__(ref, ref2=ref2, name=name, cache_size=cache_size)

test

test(engine: Engine) -> TestResult

Source code in src/datajudge/constraints/stats.py

def test(self, engine: sa.engine.Engine) -> TestResult:
    if self._ref2 is None:
        raise ValueError("KolmogorovSmirnov2Sample requires ref2.")
    (
        d_statistic,
        p_value,
        n_samples,
        m_samples,
        selections,
    ) = self._calculate_statistic(
        engine,
        self._ref,
        self._ref2,
    )
    result = self._check_acceptance(
        d_statistic, n_samples, m_samples, self._significance_level
    )

    assertion_text = (
        f"Null hypothesis (H0) for the 2-sample Kolmogorov-Smirnov test was rejected, i.e., "
        f"the two samples ({self._ref} and {self._target_prefix}) "
        f"do not originate from the same distribution. "
        f"The test results are d={d_statistic}"
    )
    if p_value is not None:
        assertion_text += f" and {p_value=}"
    assertion_text += "."

    if selections:
        queries = [
            str(selection.compile(engine, compile_kwargs={"literal_binds": True}))
            for selection in selections
        ]

    if not result:
        return TestResult.failure(
            assertion_text,
            self.get_description(),
            queries,
        )

    return TestResult.success()

uniques

CategoricalBoundConstraint

CategoricalBoundConstraint(ref: DataReference, distribution: dict[_T, tuple[float, float]], default_bounds: tuple[float, float] = (0, 0), name: str | None = None, cache_size=None, max_relative_violations: float = 0, **kwargs)

Bases: Constraint

Constraint that checks if the share of specific values in a column falls within predefined bounds.

It compares the actual distribution of values in a DataSource column with a target distribution, supplied as a dictionary.

Example use cases include testing for consistency in columns with expected categorical values or ensuring that the distribution of values in a column adheres to a certain criterion.

PARAMETER	DESCRIPTION
ref	A reference to the column in the data source. TYPE: DataReference
distribution	A dictionary with unique values as keys and tuples of minimum and maximum allowed shares as values. TYPE: dict[_T, tuple[float, float]]
default_bounds	A tuple specifying the minimum and maximum bounds for values not explicitly outlined in the target distribution dictionary. TYPE: tuple[float, float] DEFAULT: (0, 0)
name	An optional name for the constraint. TYPE: str | None DEFAULT: None
max_relative_violations	A tolerance threshold (0 to 1) for the proportion of elements in the data that can violate the bound constraints without triggering the constraint violation. TYPE: float DEFAULT: 0

Source code in src/datajudge/constraints/uniques.py

def __init__(
    self,
    ref: DataReference,
    distribution: dict[_T, tuple[float, float]],
    default_bounds: tuple[float, float] = (0, 0),
    name: str | None = None,
    cache_size=None,
    max_relative_violations: float = 0,
    **kwargs,
):
    self._default_bounds = default_bounds
    self._max_relative_violations = max_relative_violations
    super().__init__(
        ref,
        ref_value=distribution,
        name=name,
        cache_size=cache_size,
        **kwargs,
    )

NUniques

NUniques(ref: DataReference, *, ref2: DataReference | None = None, n_uniques: int | None = None, name: str | None = None, cache_size=None)

Bases: Constraint, ABC

Source code in src/datajudge/constraints/uniques.py

def __init__(
    self,
    ref: DataReference,
    *,
    ref2: DataReference | None = None,
    n_uniques: int | None = None,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=n_uniques,
        name=name,
        cache_size=cache_size,
    )

NUniquesEquality

NUniquesEquality(ref: DataReference, *, ref2: DataReference | None = None, n_uniques: int | None = None, name: str | None = None, cache_size=None)

Bases: NUniques

Source code in src/datajudge/constraints/uniques.py

def __init__(
    self,
    ref: DataReference,
    *,
    ref2: DataReference | None = None,
    n_uniques: int | None = None,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=n_uniques,
        name=name,
        cache_size=cache_size,
    )

NUniquesMaxGain

NUniquesMaxGain(ref: DataReference, ref2: DataReference, max_relative_gain_getter: _ToleranceGetter, name: str | None = None, cache_size=None)

Bases: NUniques

Source code in src/datajudge/constraints/uniques.py

def __init__(
    self,
    ref: DataReference,
    ref2: DataReference,
    max_relative_gain_getter: _ToleranceGetter,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(ref, ref2=ref2, name=name, cache_size=cache_size)
    self.max_relative_gain_getter = max_relative_gain_getter

max_relative_gain_getter instance-attribute

max_relative_gain_getter = max_relative_gain_getter

test

test(engine: Engine) -> TestResult

Source code in src/datajudge/constraints/uniques.py

def test(self, engine: sa.engine.Engine) -> TestResult:
    self.max_relative_gain = self.max_relative_gain_getter(engine)
    return super().test(engine)

NUniquesMaxLoss

NUniquesMaxLoss(ref: DataReference, ref2: DataReference, max_relative_loss_getter: _ToleranceGetter, name: str | None = None, cache_size=None)

Bases: NUniques

Source code in src/datajudge/constraints/uniques.py

def __init__(
    self,
    ref: DataReference,
    ref2: DataReference,
    max_relative_loss_getter: _ToleranceGetter,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(ref, ref2=ref2, name=name, cache_size=cache_size)
    self.max_relative_loss_getter = max_relative_loss_getter

max_relative_loss_getter instance-attribute

max_relative_loss_getter = max_relative_loss_getter

test

test(engine: Engine) -> TestResult

Source code in src/datajudge/constraints/uniques.py

def test(self, engine: sa.engine.Engine) -> TestResult:
    self.max_relative_loss = self.max_relative_loss_getter(engine)
    return super().test(engine)

Uniques

Uniques(ref: DataReference, name: str | None = None, cache_size=None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, *, ref2: DataReference | None = None, uniques: Collection | None = None, filter_func: Callable[[list[_T]], list[_T]] | None = None, map_func: Callable[[_T], _T] | None = None, reduce_func: Callable[[Collection], Collection] | None = None, max_relative_violations=0, compare_distinct=False)

Bases: Constraint, ABC

Uniques is an abstract class for comparisons between unique values of a column and a reference.

The Uniques constraint asserts if the values contained in a column of a DataSource are part of a reference set of expected values - either externally supplied through parameter uniques or obtained from another DataSource.

Null values in the columns columns are ignored. To assert the non-existence of them use the add_null_absence_constraint helper method for WithinRequirement. By default, the null filtering does not trigger if multiple columns are fetched at once. It can be configured in more detail by supplying a custom filter_func function. Some exemplary implementations are available as filternull_element, filternull_never, filternull_element_or_tuple_all, filternull_element_or_tuple_any. Passing None as the argument is equivalent to filternull_element but triggers a warning. The current default of filternull_element Cause (possibly often unintended) changes in behavior when the users adds a second column (filtering no longer can trigger at all). The default will be changed to filternull_element_or_tuple_all in future versions. To silence the warning, set filter_func explicitly.

There are two ways to do some post processing of the data obtained from the database by providing a function to be executed. In general, no postprocessing is needed, but there are some cases where it's the only thing to do. For example, with text values that have some structure.

One is map_func, it'll be executed over each obtained 'unique' value. This is a very local operation.

If map_func is provided, it'll be executed over each obtained 'unique' value.

The second one is reduce_func which will take the whole data retrieved and can perform global processing. If it is provided, it gets applied after the function given in map_func is finished. The output of this function has to be an iterable (eager or lazy) of the same type as the type of the values of the column (in their Python equivalent).

Furthermore, the max_relative_violations parameter can be used to set a tolerance threshold for the proportion of elements in the data that can violate the constraint (default: 0). Setting this argument is currently not supported for UniquesEquality.

For UniquesSubset, by default, the number of occurrences affects the computed fraction of violations. To disable this weighting, set compare_distinct=True. This argument does not have an effect on the test results for other Uniques constraints, or if max_relative_violations is 0.

By default, the assertion messages make use of sets, thus, they may differ from run to run despite the exact same situation being present, and can have an arbitrary length. To enforce a reproducible, limited output via (e.g.) sorting and slicing, set output_processors to a callable or a list of callables. By default, only the first 100 elements are displayed (output_processor_limit).

Each callable takes in two collections, and returns modified (e.g. sorted) versions of them. In most cases, the second argument is simply None, but for UniquesSubset it is the counts of each of the elements. The suggested functions are output_processor_sort and output_processor_limit - see their respective docstrings for details.

One use is of this constraint is to test for consistency in columns with expected categorical values.

Source code in src/datajudge/constraints/uniques.py

def __init__(
    self,
    ref: DataReference,
    name: str | None = None,
    cache_size=None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    *,
    ref2: DataReference | None = None,
    uniques: Collection | None = None,
    filter_func: Callable[[list[_T]], list[_T]] | None = None,
    map_func: Callable[[_T], _T] | None = None,
    reduce_func: Callable[[Collection], Collection] | None = None,
    max_relative_violations=0,
    compare_distinct=False,
):
    ref_value: tuple[Collection, list] | None
    ref_value = (uniques, []) if uniques else None
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=ref_value,
        name=name,
        cache_size=cache_size,
        output_processors=output_processors,
    )

    if filter_func is None:
        warnings.warn("Using deprecated default null filter function.")
        filter_func = filternull_element

    self._filter_func = filter_func
    self._local_func = map_func
    self._global_func = reduce_func
    self._max_relative_violations = max_relative_violations
    self._compare_distinct = compare_distinct

UniquesEquality

UniquesEquality(args, name: str | None = None, cache_size=None, **kwargs)

Bases: Uniques

Source code in src/datajudge/constraints/uniques.py

def __init__(self, args, name: str | None = None, cache_size=None, **kwargs):
    if kwargs.get("max_relative_violations"):
        raise RuntimeError(
            "max_relative_violations is not supported for UniquesEquality."
        )
    if kwargs.get("compare_distinct"):
        raise RuntimeError("compare_distinct is not supported for UniquesEquality.")
    super().__init__(args, name=name, **kwargs)

UniquesSubset

UniquesSubset(ref: DataReference, name: str | None = None, cache_size=None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, *, ref2: DataReference | None = None, uniques: Collection | None = None, filter_func: Callable[[list[_T]], list[_T]] | None = None, map_func: Callable[[_T], _T] | None = None, reduce_func: Callable[[Collection], Collection] | None = None, max_relative_violations=0, compare_distinct=False)

Bases: Uniques

Source code in src/datajudge/constraints/uniques.py

def __init__(
    self,
    ref: DataReference,
    name: str | None = None,
    cache_size=None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    *,
    ref2: DataReference | None = None,
    uniques: Collection | None = None,
    filter_func: Callable[[list[_T]], list[_T]] | None = None,
    map_func: Callable[[_T], _T] | None = None,
    reduce_func: Callable[[Collection], Collection] | None = None,
    max_relative_violations=0,
    compare_distinct=False,
):
    ref_value: tuple[Collection, list] | None
    ref_value = (uniques, []) if uniques else None
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=ref_value,
        name=name,
        cache_size=cache_size,
        output_processors=output_processors,
    )

    if filter_func is None:
        warnings.warn("Using deprecated default null filter function.")
        filter_func = filternull_element

    self._filter_func = filter_func
    self._local_func = map_func
    self._global_func = reduce_func
    self._max_relative_violations = max_relative_violations
    self._compare_distinct = compare_distinct

UniquesSuperset

UniquesSuperset(args, name: str | None = None, cache_size=None, **kwargs)

Bases: Uniques

Source code in src/datajudge/constraints/uniques.py

def __init__(self, args, name: str | None = None, cache_size=None, **kwargs):
    if kwargs.get("compare_distinct"):
        raise RuntimeError("compare_distinct is not supported for UniquesSuperset.")
    super().__init__(args, name=name, **kwargs)

varchar

VarCharMaxLength

VarCharMaxLength(ref: DataReference, *, ref2: DataReference | None = None, max_length: int | None = None, name: str | None = None, cache_size=None)

Bases: Constraint

Source code in src/datajudge/constraints/varchar.py

def __init__(
    self,
    ref: DataReference,
    *,
    ref2: DataReference | None = None,
    max_length: int | None = None,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=max_length,
        name=name,
        cache_size=cache_size,
    )

VarCharMinLength

VarCharMinLength(ref, *, ref2: DataReference | None = None, min_length: int | None = None, name: str | None = None, cache_size=None)

Bases: Constraint

Source code in src/datajudge/constraints/varchar.py

def __init__(
    self,
    ref,
    *,
    ref2: DataReference | None = None,
    min_length: int | None = None,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(
        ref,
        ref2=ref2,
        ref_value=min_length,
        name=name,
        cache_size=cache_size,
    )

VarCharRegex

VarCharRegex(ref: DataReference, regex: str, allow_none: bool = False, relative_tolerance: float = 0.0, aggregated: bool = True, n_counterexamples: int = 5, name: str | None = None, cache_size=None)

Bases: Constraint

Source code in src/datajudge/constraints/varchar.py

def __init__(
    self,
    ref: DataReference,
    regex: str,
    allow_none: bool = False,
    relative_tolerance: float = 0.0,
    aggregated: bool = True,
    n_counterexamples: int = 5,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(ref, ref_value=regex, name=name, cache_size=cache_size)
    self._allow_none = allow_none
    self._relative_tolerance = relative_tolerance
    self._aggregated = aggregated
    self._n_counterexamples = n_counterexamples

test

test(engine: Engine) -> TestResult

Source code in src/datajudge/constraints/varchar.py

def test(self, engine: sa.engine.Engine) -> TestResult:
    uniques_counter, selections = db_access.get_uniques(engine, self._ref)
    self.factual_selections = selections
    if not self._allow_none and uniques_counter.get(None):
        return TestResult.failure(
            "The column contains a None value when it's not allowed. "
            "To ignore None values, please use `allow_none=True` option."
        )
    elif None in uniques_counter:
        uniques_counter.pop(None)

    uniques_factual = list(uniques_counter.keys())
    if not self._ref_value:
        return TestResult.failure("No regex pattern given")

    pattern = re.compile(self._ref_value)
    uniques_mismatching = {
        x
        for x in uniques_factual
        if not pattern.match(x)  # type: ignore
    }

    if self._aggregated:
        n_violations = len(uniques_mismatching)
        n_total = len(uniques_factual)
    else:
        n_violations = sum(uniques_counter[key] for key in uniques_mismatching)
        n_total = sum(count for _, count in uniques_counter.items())

    n_relative_violations = n_violations / n_total

    if self._n_counterexamples == -1:
        counterexamples = list(uniques_mismatching)
    else:
        counterexamples = list(
            itertools.islice(uniques_mismatching, self._n_counterexamples)
        )

    counterexample_string = (
        (f"Some counterexamples consist of the following: {counterexamples}. ")
        if counterexamples and len(counterexamples) > 0
        else ""
    )

    if n_relative_violations > self._relative_tolerance:
        assertion_text = (
            f"{self._ref} "
            f"breaks regex '{self._ref_value}' in {n_relative_violations} > "
            f"{self._relative_tolerance} of the cases. "
            f"In absolute terms, {n_violations} of the {n_total} samples violated the regex. "
            f"{counterexample_string}{self._condition_string}"
        )
        return TestResult.failure(assertion_text)
    return TestResult.success()

VarCharRegexDb

VarCharRegexDb(ref: DataReference, regex: str, relative_tolerance: float = 0.0, aggregated: bool = True, n_counterexamples: int = 5, name: str | None = None, cache_size=None)

Bases: Constraint

Source code in src/datajudge/constraints/varchar.py

def __init__(
    self,
    ref: DataReference,
    regex: str,
    relative_tolerance: float = 0.0,
    aggregated: bool = True,
    n_counterexamples: int = 5,
    name: str | None = None,
    cache_size=None,
):
    super().__init__(
        ref,
        ref_value=relative_tolerance,
        name=name,
        cache_size=cache_size,
    )
    self._regex = regex
    self._aggregated = aggregated
    self._n_counterexamples = n_counterexamples

data_source

DataSource

Bases: ABC

ExpressionDataSource

ExpressionDataSource(expression: FromClause | Select, name: str)

Bases: DataSource

A DataSource based on a sqlalchemy expression.

Source code in src/datajudge/data_source.py

def __init__(
    self,
    expression: sa.sql.selectable.FromClause | sa.sql.selectable.Select,
    name: str,
):
    self._expression = expression
    self.name = name

name instance-attribute

name = name

RawQueryDataSource

RawQueryDataSource(query_string: str, name: str, columns: list[str] | None = None)

Bases: DataSource

A DataSource based on a SQL query as a string.

Source code in src/datajudge/data_source.py

def __init__(self, query_string: str, name: str, columns: list[str] | None = None):
    self._query_string = query_string
    self.name = name
    self._columns = columns
    wrapped_query = f"({query_string}) as t"
    if columns is not None and len(columns) > 0:
        subquery = (
            sa.text(query_string)
            .columns(*[sa.column(column_name) for column_name in columns])
            .subquery()
        )
        self.clause = subquery
    else:
        wrapped_query = f"({query_string}) as t"
        self.clause = sa.select("*").select_from(sa.text(wrapped_query)).alias()

clause instance-attribute

clause = subquery

name instance-attribute

name = name

TableDataSource

TableDataSource(db_name: str, table_name: str, schema_name: str | None = None)

Bases: DataSource

A DataSource based on a table.

Source code in src/datajudge/data_source.py

def __init__(
    self,
    db_name: str,
    table_name: str,
    schema_name: str | None = None,
):
    self._db_name = db_name
    self._table_name = table_name
    self._schema_name = schema_name

formatter

AnsiColorFormatter

AnsiColorFormatter()

Bases: Formatter

Source code in src/datajudge/formatter.py

def __init__(self):
    super().__init__()
    just_fix_windows_console()

Formatter

Formatter()

Bases: ABC

Source code in src/datajudge/formatter.py

def __init__(self):
    self._known_bb_pattern = re.compile(_STYLING_CODES)

fmt_str

fmt_str(string: str) -> str

Source code in src/datajudge/formatter.py

def fmt_str(self, string: str) -> str:
    # Replace codes with platform specific styling
    string = self._known_bb_pattern.sub(
        lambda m: self._apply_formatting(m.group(1), m.group(2)), string
    )

    return string

pytest_integration

collect_data_tests

collect_data_tests(requirements: Iterable[Requirement])

Make a Pytest test case that checks all requirements.

Returns a function named test_constraint that is parametrized over all constraints in requirements. The function requires a datajudge_engine fixture that is a SQLAlchemy engine to be available.

Source code in src/datajudge/pytest_integration.py

def collect_data_tests(requirements: Iterable[Requirement]):
    """Make a Pytest test case that checks all `requirements`.

    Returns a function named `test_constraint` that is parametrized over all
    constraints in `requirements`. The function requires a `datajudge_engine`
    fixture that is a SQLAlchemy engine to be available.
    """
    all_constraints = [
        constraint for requirement in requirements for constraint in requirement
    ]

    @pytest.mark.parametrize(
        "constraint", all_constraints, ids=Constraint.get_description
    )
    def test_constraint(constraint, datajudge_engine, pytestconfig):
        # apply patches that fix sqlalchemy issues
        formatter = get_formatter(pytestconfig)
        apply_patches(datajudge_engine)
        test_result = constraint.test(datajudge_engine)
        assert test_result.outcome, test_result.formatted_failure_message(formatter)

    return test_constraint

get_formatter

get_formatter(pytestconfig)

Source code in src/datajudge/pytest_integration.py

def get_formatter(pytestconfig):
    color = pytestconfig.getoption("color")

    if color == "yes" or color == "auto":
        # before pytest-html < 4.0.0
        # styling in assertion messages was not formatted correctly
        # so in this case we use the default formatter
        # in pytest-html >= 4.0.0 the styling gets stripped or
        # translated to ANSI codes, depending if ansi2html is installed
        if pytestconfig.getoption("htmlpath", False):
            try:
                import pytest_html

                if Version(pytest_html.__version__).major >= 4:
                    return AnsiColorFormatter()
            finally:
                return Formatter()

        return AnsiColorFormatter()
    else:
        return Formatter()

requirements

BetweenRequirement

BetweenRequirement(data_source: DataSource, data_source2: DataSource, date_column: str | None = None, date_column2: str | None = None)

Bases: Requirement

Source code in src/datajudge/requirements.py

def __init__(
    self,
    data_source: DataSource,
    data_source2: DataSource,
    date_column: str | None = None,
    date_column2: str | None = None,
):
    self._data_source = data_source
    self._data_source2 = data_source2
    self._ref = DataReference(self._data_source)
    self._ref2 = DataReference(self._data_source2)
    self._date_column = date_column
    self._date_column2 = date_column2
    super().__init__()

add_column_subset_constraint

add_column_subset_constraint(name: str | None = None, cache_size=None) -> None

Columns of first table are subset of second table.

Source code in src/datajudge/requirements.py

def add_column_subset_constraint(
    self, name: str | None = None, cache_size=None
) -> None:
    """Columns of first table are subset of second table."""
    self._constraints.append(
        column_constraints.ColumnSubset(
            self._ref, ref2=self._ref2, name=name, cache_size=cache_size
        )
    )

add_column_superset_constraint

add_column_superset_constraint(name: str | None = None, cache_size=None) -> None

Columns of first table are superset of columns of second table.

Source code in src/datajudge/requirements.py

def add_column_superset_constraint(
    self, name: str | None = None, cache_size=None
) -> None:
    """Columns of first table are superset of columns of second table."""
    self._constraints.append(
        column_constraints.ColumnSuperset(
            self._ref, ref2=self._ref2, name=name, cache_size=cache_size
        )
    )

add_column_type_constraint

add_column_type_constraint(column1: str, column2: str, name: str | None = None, cache_size=None) -> None

Check that the columns have the same type.

Source code in src/datajudge/requirements.py

def add_column_type_constraint(
    self,
    column1: str,
    column2: str,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Check that the columns have the same type."""
    ref1 = DataReference(self._data_source, [column1])
    ref2 = DataReference(self._data_source2, [column2])
    self._constraints.append(
        column_constraints.ColumnType(
            ref1, ref2=ref2, name=name, cache_size=cache_size
        )
    )

add_date_max_constraint

add_date_max_constraint(column1: str, column2: str, use_upper_bound_reference: bool = True, column_type: str = 'date', condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Compare date max of first table to date max of second table.

The used columns of both tables need to be of the same type.

For more information on column_type values, see add_column_type_constraint.

If use_upper_bound_reference, the max of the first table has to be smaller or equal to the max of the second table. If not use_upper_bound_reference, the max of the first table has to be greater or equal to the max of the second table.

Source code in src/datajudge/requirements.py

def add_date_max_constraint(
    self,
    column1: str,
    column2: str,
    use_upper_bound_reference: bool = True,
    column_type: str = "date",
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Compare date max of first table to date max of second table.

    The used columns of both tables need to be of the same type.

    For more information on ``column_type`` values, see ``add_column_type_constraint``.

    If ``use_upper_bound_reference``, the max of the first table has to be
    smaller or equal to the max of the second table.
    If not ``use_upper_bound_reference``, the max of the first table has to
    be greater or equal to the max of the second table.
    """
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        date_constraints.DateMax(
            ref,
            ref2=ref2,
            use_upper_bound_reference=use_upper_bound_reference,
            column_type=column_type,
            name=name,
            cache_size=cache_size,
        )
    )

add_date_min_constraint

add_date_min_constraint(column1: str, column2: str, use_lower_bound_reference: bool = True, column_type: str = 'date', condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Ensure date min of first table is greater or equal date min of second table.

The used columns of both tables need to be of the same type.

For more information on column_type values, see add_column_type_constraint.

If use_lower_bound_reference, the min of the first table has to be greater or equal to the min of the second table. If not use_upper_bound_reference, the min of the first table has to be smaller or equal to the min of the second table.

Source code in src/datajudge/requirements.py

def add_date_min_constraint(
    self,
    column1: str,
    column2: str,
    use_lower_bound_reference: bool = True,
    column_type: str = "date",
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Ensure date min of first table is greater or equal date min of second table.

    The used columns of both tables need to be of the same type.

    For more information on ``column_type`` values, see ``add_column_type_constraint``.

    If ``use_lower_bound_reference``, the min of the first table has to be
    greater or equal to the min of the second table.
    If not ``use_upper_bound_reference``, the min of the first table has to
    be smaller or equal to the min of the second table.
    """
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        date_constraints.DateMin(
            ref,
            ref2=ref2,
            use_lower_bound_reference=use_lower_bound_reference,
            column_type=column_type,
            name=name,
            cache_size=cache_size,
        )
    )

add_ks_2sample_constraint

add_ks_2sample_constraint(column1: str, column2: str, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, significance_level: float = 0.05, cache_size=None) -> None

Apply the so-called two-sample Kolmogorov-Smirnov test to the distributions of the two given columns.

The constraint is fulfilled, when the resulting p-value of the test is higher than the significance level (default is 0.05, i.e., 5%). The significance_level must be a value between 0.0 and 1.0.

Source code in src/datajudge/requirements.py

def add_ks_2sample_constraint(
    self,
    column1: str,
    column2: str,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    significance_level: float = 0.05,
    cache_size=None,
) -> None:
    """
    Apply the so-called two-sample Kolmogorov-Smirnov test to the distributions of the two given columns.

    The constraint is fulfilled, when the resulting p-value of the test is higher than the significance level
    (default is 0.05, i.e., 5%).
    The significance_level must be a value between 0.0 and 1.0.
    """
    if not column1 or not column2:
        raise ValueError(
            "Column names have to be given for this test's functionality."
        )

    if significance_level <= 0.0 or significance_level > 1.0:
        raise ValueError(
            "The requested significance level has to be in ``(0.0, 1.0]``. Default is 0.05."
        )

    ref = DataReference(self._data_source, [column1], condition=condition1)
    ref2 = DataReference(self._data_source2, [column2], condition=condition2)
    self._constraints.append(
        stats_constraints.KolmogorovSmirnov2Sample(
            ref,
            ref2,
            significance_level,
            name=name,
            cache_size=cache_size,
        )
    )

add_max_null_fraction_constraint

add_max_null_fraction_constraint(column1: str, column2: str, max_relative_deviation: float, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the fraction of NULL values of one is at most that of the other.

Given that column2's underlying data has a fraction q of NULL values, the max_relative_deviation parameter allows column1's underlying data to have a fraction (1 + max_relative_deviation) * q of NULL values.

Source code in src/datajudge/requirements.py

def add_max_null_fraction_constraint(
    self,
    column1: str,
    column2: str,
    max_relative_deviation: float,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Assert that the fraction of ``NULL`` values of one is at most that of the other.

    Given that ``column2``'s underlying data has a fraction ``q`` of ``NULL`` values, the
    ``max_relative_deviation`` parameter allows ``column1``'s underlying data to have a
    fraction ``(1 + max_relative_deviation) * q`` of ``NULL`` values.
    """  # noqa: D301
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        miscs_constraints.MaxNullFraction(
            ref,
            ref2=ref2,
            max_relative_deviation=max_relative_deviation,
            cache_size=cache_size,
        )
    )

add_n_rows_equality_constraint

add_n_rows_equality_constraint(condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_n_rows_equality_constraint(
    self,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    ref = DataReference(self._data_source, condition=condition1)
    ref2 = DataReference(self._data_source2, condition=condition2)
    self._constraints.append(
        nrows_constraints.NRowsEquality(
            ref, ref2=ref2, name=name, cache_size=cache_size
        )
    )

add_n_rows_max_gain_constraint

add_n_rows_max_gain_constraint(constant_max_relative_gain: float | None = None, date_range_gain_deviation: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the number of rows hasn't grown by more than expected.

In particular, assert that

\[n^{rows}_1 \leq n^{rows}_2 \cdot (1 + \text{cmrg})\]

See readme for more information on constant_max_relative_gain.

Source code in src/datajudge/requirements.py

def add_n_rows_max_gain_constraint(
    self,
    constant_max_relative_gain: float | None = None,
    date_range_gain_deviation: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    r"""Assert that the number of rows hasn't grown by more than expected.

    In particular, assert that

    $$n^{rows}_1 \leq n^{rows}_2 \cdot (1 + \text{cmrg})$$

    See readme for more information on ``constant_max_relative_gain``.
    """
    max_relative_gain_getter = self._get_deviation_getter(
        constant_max_relative_gain, date_range_gain_deviation
    )
    ref = DataReference(self._data_source, condition=condition1)
    ref2 = DataReference(self._data_source2, condition=condition2)
    self._constraints.append(
        nrows_constraints.NRowsMaxGain(
            ref,
            ref2,
            max_relative_gain_getter,
            name=name,
            cache_size=cache_size,
        )
    )

add_n_rows_max_loss_constraint

add_n_rows_max_loss_constraint(constant_max_relative_loss: float | None = None, date_range_loss_deviation: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the number of rows hasn't decreased too much.

In particular, assert that

\[n^{rows}_1 \geq n^{rows}_2 \cdot (1 - \text{cmrl})\]

See readme for more information on constant_max_relative_loss.

Source code in src/datajudge/requirements.py

def add_n_rows_max_loss_constraint(
    self,
    constant_max_relative_loss: float | None = None,
    date_range_loss_deviation: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    r"""Assert that the number of rows hasn't decreased too much.

    In particular, assert that

    $$n^{rows}_1 \geq n^{rows}_2 \cdot (1 - \text{cmrl})$$

    See readme for more information on ``constant_max_relative_loss``.
    """
    max_relative_loss_getter = self._get_deviation_getter(
        constant_max_relative_loss, date_range_loss_deviation
    )
    ref = DataReference(self._data_source, condition=condition1)
    ref2 = DataReference(self._data_source2, condition=condition2)
    self._constraints.append(
        nrows_constraints.NRowsMaxLoss(
            ref,
            ref2,
            max_relative_loss_getter,
            name=name,
            cache_size=cache_size,
        )
    )

add_n_rows_min_gain_constraint

add_n_rows_min_gain_constraint(constant_min_relative_gain: float | None = None, date_range_gain_deviation: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the number of rows hasn't grown less than expected.

In particular, assert that

\[n^{rows}_1 \geq n^{rows}_2 \cdot (1 + \text{cmrg})\]

See readme for more information on constant_min_relative_gain.

Source code in src/datajudge/requirements.py

def add_n_rows_min_gain_constraint(
    self,
    constant_min_relative_gain: float | None = None,
    date_range_gain_deviation: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    r"""Assert that the number of rows hasn't grown less than expected.

    In particular, assert that

    $$n^{rows}_1 \geq n^{rows}_2 \cdot (1 + \text{cmrg})$$

    See readme for more information on ``constant_min_relative_gain``.
    """
    min_relative_gain_getter = self._get_deviation_getter(
        constant_min_relative_gain, date_range_gain_deviation
    )
    ref = DataReference(self._data_source, condition=condition1)
    ref2 = DataReference(self._data_source2, condition=condition2)
    self._constraints.append(
        nrows_constraints.NRowsMinGain(
            ref,
            ref2,
            min_relative_gain_getter,
            name=name,
            cache_size=cache_size,
        )
    )

add_n_uniques_equality_constraint

add_n_uniques_equality_constraint(columns1: list[str] | None, columns2: list[str] | None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_n_uniques_equality_constraint(
    self,
    columns1: list[str] | None,
    columns2: list[str] | None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        uniques_constraints.NUniquesEquality(
            ref, ref2=ref2, name=name, cache_size=cache_size
        )
    )

add_n_uniques_max_gain_constraint

add_n_uniques_max_gain_constraint(columns1: list[str] | None, columns2: list[str] | None, constant_max_relative_gain: float | None = None, date_range_gain_deviation: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the number of uniques hasn't grown by too much.

In particular, assert that

\[n^{uniques}_1 \leq n^{uniques}_2 \cdot (1 - \text{cmrg})\]

The number of uniques in first table are defined based on columns1, the number of uniques in second table are defined based on columns2.

See readme for more information on constant_max_relative_gain.

Source code in src/datajudge/requirements.py

def add_n_uniques_max_gain_constraint(
    self,
    columns1: list[str] | None,
    columns2: list[str] | None,
    constant_max_relative_gain: float | None = None,
    date_range_gain_deviation: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    r"""Assert that the number of uniques hasn't grown by too much.

    In particular, assert that

    $$n^{uniques}_1 \leq n^{uniques}_2 \cdot (1 - \text{cmrg})$$

    The number of uniques in first table are defined based on ``columns1``, the
    number of uniques in second table are defined based on ``columns2``.

    See readme for more information on ``constant_max_relative_gain``.
    """
    max_relative_gain_getter = self._get_deviation_getter(
        constant_max_relative_gain, date_range_gain_deviation
    )
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        uniques_constraints.NUniquesMaxGain(
            ref,
            ref2,
            max_relative_gain_getter,
            name=name,
            cache_size=cache_size,
        )
    )

add_n_uniques_max_loss_constraint

add_n_uniques_max_loss_constraint(columns1: list[str] | None, columns2: list[str] | None, constant_max_relative_loss: float | None = None, date_range_loss_deviation: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the number of uniques hasn't decreased too much.

In particular, assert that

\[n^{uniques}_1 \geq n^{uniques}_2 \cdot (1 - \text{cmrl})\]

The number of uniques in first table are defined based on columns1, the number of uniques in second table are defined based on columns2.

See readme for more information on constant_max_relative_loss.

Source code in src/datajudge/requirements.py

def add_n_uniques_max_loss_constraint(
    self,
    columns1: list[str] | None,
    columns2: list[str] | None,
    constant_max_relative_loss: float | None = None,
    date_range_loss_deviation: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    r"""Assert that the number of uniques hasn't decreased too much.

    In particular, assert that

    $$n^{uniques}_1 \geq n^{uniques}_2 \cdot (1 - \text{cmrl})$$

    The number of uniques in first table are defined based on ``columns1``, the
    number of uniques in second table are defined based on ``columns2``.

    See readme for more information on ``constant_max_relative_loss``.
    """
    max_relative_loss_getter = self._get_deviation_getter(
        constant_max_relative_loss, date_range_loss_deviation
    )
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        uniques_constraints.NUniquesMaxLoss(
            ref,
            ref2,
            max_relative_loss_getter,
            name=name,
            cache_size=cache_size,
        )
    )

add_numeric_max_constraint

add_numeric_max_constraint(column1: str, column2: str, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_numeric_max_constraint(
    self,
    column1: str,
    column2: str,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        numeric_constraints.NumericMax(
            ref, ref2=ref2, name=name, cache_size=cache_size
        )
    )

add_numeric_mean_constraint

add_numeric_mean_constraint(column1: str, column2: str, max_absolute_deviation: float, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_numeric_mean_constraint(
    self,
    column1: str,
    column2: str,
    max_absolute_deviation: float,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        numeric_constraints.NumericMean(
            ref,
            max_absolute_deviation,
            ref2=ref2,
            name=name,
            cache_size=cache_size,
        )
    )

add_numeric_min_constraint

add_numeric_min_constraint(column1: str, column2: str, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_numeric_min_constraint(
    self,
    column1: str,
    column2: str,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        numeric_constraints.NumericMin(
            ref, ref2=ref2, name=name, cache_size=cache_size
        )
    )

add_numeric_percentile_constraint

add_numeric_percentile_constraint(column1: str, column2: str, percentage: float, max_absolute_deviation: float | None = None, max_relative_deviation: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the percentage-th percentile is approximately equal.

The percentile is defined as the smallest value present in column1 / column2 for which percentage % of the values in column1 / column2 are less or equal. NULL values are ignored.

Hence, if percentage is less than the inverse of the number of non-NULL rows, None is received as the percentage-th percentile.

percentage is expected to be provided in percent. The median, for example, would correspond to percentage=50.

At least one of max_absolute_deviation and max_relative_deviation must be provided.

Source code in src/datajudge/requirements.py

def add_numeric_percentile_constraint(
    self,
    column1: str,
    column2: str,
    percentage: float,
    max_absolute_deviation: float | None = None,
    max_relative_deviation: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Assert that the ``percentage``-th percentile is approximately equal.

    The percentile is defined as the smallest value present in ``column1`` / ``column2``
    for which ``percentage`` % of the values in ``column1`` / ``column2`` are
    less or equal. ``NULL`` values are ignored.

    Hence, if ``percentage`` is less than the inverse of the number of non-``NULL``
    rows, ``None`` is received as the ``percentage``-th percentile.

    ``percentage`` is expected to be provided in percent. The median, for example,
    would correspond to ``percentage=50``.

    At least one of ``max_absolute_deviation`` and ``max_relative_deviation`` must
    be provided.
    """
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        numeric_constraints.NumericPercentile(
            ref,
            percentage=percentage,
            max_absolute_deviation=max_absolute_deviation,
            max_relative_deviation=max_relative_deviation,
            ref2=ref2,
            name=name,
            cache_size=cache_size,
        )
    )

add_row_equality_constraint

add_row_equality_constraint(columns1: list[str] | None, columns2: list[str] | None, max_missing_fraction: float, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

At most max_missing_fraction of rows in T1 and T2 are absent in either.

In other words

\[\frac{|T1 - T2| + |T2 - T1|}{|T1 \cup T2|} \leq \text{mmf}\]

Rows from T1 are indexed in columns1, rows from T2 are indexed in columns2.

Source code in src/datajudge/requirements.py

def add_row_equality_constraint(
    self,
    columns1: list[str] | None,
    columns2: list[str] | None,
    max_missing_fraction: float,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """At most ``max_missing_fraction`` of rows in T1 and T2 are absent in either.

    In other words

    $$\\frac{|T1 - T2| + |T2 - T1|}{|T1 \\cup T2|} \\leq \\text{mmf}$$

    Rows from T1 are indexed in ``columns1``, rows from T2 are indexed in ``columns2``.
    """  # noqa: D301
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        row_constraints.RowEquality(
            ref,
            ref2,
            lambda engine: max_missing_fraction,
            name=name,
            cache_size=cache_size,
        )
    )

add_row_matching_equality_constraint

add_row_matching_equality_constraint(matching_columns1: list[str], matching_columns2: list[str], comparison_columns1: list[str], comparison_columns2: list[str], max_missing_fraction: float, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Match tables in matching_columns, compare for equality in comparison_columns.

This constraint is similar to the nature of the RowEquality constraint. Just as the latter, this constraint divides the cardinality of an intersection by the cardinality of a union. The difference lies in how the set are created. While RowEquality considers all rows of both tables, indexed in columns, RowMatchingEquality considers only rows in both tables having values in matching_columns present in both tables. At most max_missing_fraction of such rows can be missing in the intersection.

Alternatively, this can be thought of as counting mismatches in comparison_columns after performing an inner join on matching_columns.

Source code in src/datajudge/requirements.py

def add_row_matching_equality_constraint(
    self,
    matching_columns1: list[str],
    matching_columns2: list[str],
    comparison_columns1: list[str],
    comparison_columns2: list[str],
    max_missing_fraction: float,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Match tables in matching_columns, compare for equality in comparison_columns.

    This constraint is similar to the nature of the ``RowEquality``
    constraint. Just as the latter, this constraint divides the
    cardinality of an intersection by the cardinality of a union.
    The difference lies in how the set are created. While ``RowEquality``
    considers all rows of both tables, indexed in columns,
    ``RowMatchingEquality`` considers only rows in both tables having values
    in ``matching_columns`` present in both tables. At most ``max_missing_fraction``
    of such rows can be missing in the intersection.

    Alternatively, this can be thought of as counting mismatches in
    ``comparison_columns`` after performing an inner join on ``matching_columns``.
    """
    ref = DataReference(
        self._data_source, matching_columns1 + comparison_columns1, condition1
    )
    ref2 = DataReference(
        self._data_source2, matching_columns2 + comparison_columns2, condition2
    )
    self._constraints.append(
        row_constraints.RowMatchingEquality(
            ref,
            ref2,
            matching_columns1,
            matching_columns2,
            comparison_columns1,
            comparison_columns2,
            lambda engine: max_missing_fraction,
            name=name,
            cache_size=cache_size,
        )
    )

add_row_subset_constraint

add_row_subset_constraint(columns1: list[str] | None, columns2: list[str] | None, constant_max_missing_fraction: float | None, date_range_loss_fraction: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

At most max_missing_fraction of rows in T1 are not in T2.

In other words, :math:\frac{|T1-T2|}{|T1|} \leq max_missing_fraction. Rows from T1 are indexed in columns1, rows from T2 are indexed in columns2.

In particular, the operation |T1-T2| relies on a sql EXCEPT statement. In contrast to EXCEPT ALL, this should lead to a set subtraction instead of a multiset subtraction. In other words, duplicates in T1 are treated as single occurrences.

Source code in src/datajudge/requirements.py

def add_row_subset_constraint(
    self,
    columns1: list[str] | None,
    columns2: list[str] | None,
    constant_max_missing_fraction: float | None,
    date_range_loss_fraction: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """At most ``max_missing_fraction`` of rows in T1 are not in T2.

    In other words,
    :math:`\\frac{|T1-T2|}{|T1|} \\leq` ``max_missing_fraction``.
    Rows from T1 are indexed in columns1, rows from T2 are indexed in ``columns2``.

    In particular, the operation ``|T1-T2|`` relies on a sql ``EXCEPT`` statement. In
    contrast to ``EXCEPT ALL``, this should lead to a set subtraction instead of
    a multiset subtraction. In other words, duplicates in T1 are treated as
    single occurrences.
    """  # noqa: D301
    max_missing_fraction_getter = self._get_deviation_getter(
        constant_max_missing_fraction, date_range_loss_fraction
    )
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        row_constraints.RowSubset(
            ref,
            ref2,
            max_missing_fraction_getter,
            name=name,
            cache_size=cache_size,
        )
    )

add_row_superset_constraint

add_row_superset_constraint(columns1: list[str] | None, columns2: list[str] | None, constant_max_missing_fraction: float, date_range_loss_fraction: float | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

At most max_missing_fraction of rows in T2 are not in T1.

In other words, \(\frac{|T2-T1|}{|T2|} \leq\) max_missing_fraction. Rows from T1 are indexed in columns1, rows from T2 are indexed in columns2.

Source code in src/datajudge/requirements.py

def add_row_superset_constraint(
    self,
    columns1: list[str] | None,
    columns2: list[str] | None,
    constant_max_missing_fraction: float,
    date_range_loss_fraction: float | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """At most ``max_missing_fraction`` of rows in T2 are not in T1.

    In other words, $\\frac{|T2-T1|}{|T2|} \\leq$ ``max_missing_fraction``.
    Rows from T1 are indexed in ``columns1``, rows from T2 are indexed in
    ``columns2``.
    """  # noqa: D301
    max_missing_fraction_getter = self._get_deviation_getter(
        constant_max_missing_fraction, date_range_loss_fraction
    )
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        row_constraints.RowSuperset(
            ref,
            ref2,
            max_missing_fraction_getter,
            name=name,
            cache_size=cache_size,
        )
    )

add_uniques_equality_constraint

add_uniques_equality_constraint(columns1: list[str], columns2: list[str], filter_func: Callable[[list[_T]], list[_T]] | None = None, map_func: Callable[[_T], _T] | None = None, reduce_func: Callable[[Collection], Collection] | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Check if the data's unique values in given columns are equal.

The UniquesEquality constraint asserts if the values contained in a column of a DataSource's columns, are strictly the ones of another DataSource's columns.

Null values in the columns columns are ignored. To assert the non-existence of them use the add_null_absence_constraint helper method for WithinRequirement. By default, the null filtering does not trigger if multiple columns are fetched at once. It can be configured in more detail by supplying a custom filter_func function. Some exemplary implementations are available as filternull_element, filternull_never, filternull_element_or_tuple_all, filternull_element_or_tuple_any. Passing None as the argument is equivalent to filternull_element but triggers a warning. The current default of filternull_element Cause (possibly often unintended) changes in behavior when the users adds a second column (filtering no longer can trigger at all). The default will be changed to filternull_element_or_tuple_all in future versions. To silence the warning, set filter_func explicitly..

See Uniques for further parameter details on map_func, reduce_func, and output_processors.

Source code in src/datajudge/requirements.py

def add_uniques_equality_constraint(
    self,
    columns1: list[str],
    columns2: list[str],
    filter_func: Callable[[list[_T]], list[_T]] | None = None,
    map_func: Callable[[_T], _T] | None = None,
    reduce_func: Callable[[Collection], Collection] | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Check if the data's unique values in given columns are equal.

    The ``UniquesEquality`` constraint asserts if the values contained in a column
    of a ``DataSource``'s columns, are strictly the ones of another ``DataSource``'s
    columns.

    Null values in the columns ``columns`` are ignored. To assert the non-existence of them use
    the [`add_null_absence_constraint`][datajudge.requirements.WithinRequirement.add_null_absence_constraint] helper method
    for ``WithinRequirement``.
    By default, the null filtering does not trigger if multiple columns are fetched at once.
    It can be configured in more detail by supplying a custom ``filter_func`` function.
    Some exemplary implementations are available as [`filternull_element`][datajudge.utils.filternull_element],
    [`filternull_never`][datajudge.utils.filternull_never], [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all],
    [`filternull_element_or_tuple_any`][datajudge.utils.filternull_element_or_tuple_any].
    Passing ``None`` as the argument is equivalent to [`filternull_element`][datajudge.utils.filternull_element] but triggers a warning.
    The current default of [`filternull_element`][datajudge.utils.filternull_element]
    Cause (possibly often unintended) changes in behavior when the users adds a second column
    (filtering no longer can trigger at all).
    The default will be changed to [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all] in future versions.
    To silence the warning, set ``filter_func`` explicitly..

    See [`Uniques`][datajudge.constraints.uniques.Uniques] for further parameter details on ``map_func``,
    ``reduce_func``, and ``output_processors``.
    """
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        uniques_constraints.UniquesEquality(
            ref,
            ref2=ref2,
            filter_func=filter_func,
            map_func=map_func,
            reduce_func=reduce_func,
            output_processors=output_processors,
            name=name,
            cache_size=cache_size,
        )
    )

add_uniques_subset_constraint

add_uniques_subset_constraint(columns1: list[str], columns2: list[str], max_relative_violations: float = 0, filter_func: Callable[[list[_T]], list[_T]] | None = None, compare_distinct: bool = False, map_func: Callable[[_T], _T] | None = None, reduce_func: Callable[[Collection], Collection] | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, cache_size=None) -> None

Check if the given columns's unique values in are contained in reference data.

The UniquesSubset constraint asserts if the values contained in given column of a DataSource are part of the unique values of given columns of another DataSource.

Null values in the columns columns are ignored. To assert the non-existence of them use the add_null_absence_constraint helper method for WithinRequirement. By default, the null filtering does not trigger if multiple columns are fetched at once. It can be configured in more detail by supplying a custom filter_func function. Some exemplary implementations are available as filternull_element, filternull_never, filternull_element_or_tuple_all, filternull_element_or_tuple_any. Passing None as the argument is equivalent to filternull_element but triggers a warning. The current default of filternull_element Cause (possibly often unintended) changes in behavior when the users adds a second column (filtering no longer can trigger at all). The default will be changed to filternull_element_or_tuple_all in future versions. To silence the warning, set filter_func explicitly. max_relative_violations indicates what fraction of rows of the given table may have values not included in the reference set of unique values. Please note that UniquesSubset and UniquesSuperset are not symmetrical in this regard.

By default, the number of occurrences affects the computed fraction of violations. To disable this weighting, set compare_distinct=True. This argument does not have an effect on the test results for other Uniques constraints, or if max_relative_violations is 0.

See Uniques for further details on map_func, reduce_func, and output_processors.

Source code in src/datajudge/requirements.py

def add_uniques_subset_constraint(
    self,
    columns1: list[str],
    columns2: list[str],
    max_relative_violations: float = 0,
    filter_func: Callable[[list[_T]], list[_T]] | None = None,
    compare_distinct: bool = False,
    map_func: Callable[[_T], _T] | None = None,
    reduce_func: Callable[[Collection], Collection] | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    cache_size=None,
) -> None:
    """Check if the given columns's unique values in are contained in reference data.

    The ``UniquesSubset`` constraint asserts if the values contained in given column of
    a ``DataSource`` are part of the unique values of given columns of another
    ``DataSource``.

    Null values in the columns ``columns`` are ignored. To assert the non-existence of them use
    the [`add_null_absence_constraint`][datajudge.requirements.WithinRequirement.add_null_absence_constraint] helper method
    for ``WithinRequirement``.
    By default, the null filtering does not trigger if multiple columns are fetched at once.
    It can be configured in more detail by supplying a custom ``filter_func`` function.
    Some exemplary implementations are available as [`filternull_element`][datajudge.utils.filternull_element],
    [`filternull_never`][datajudge.utils.filternull_never], [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all],
    [`filternull_element_or_tuple_any`][datajudge.utils.filternull_element_or_tuple_any].
    Passing ``None`` as the argument is equivalent to [`filternull_element`][datajudge.utils.filternull_element] but triggers a warning.
    The current default of [`filternull_element`][datajudge.utils.filternull_element]
    Cause (possibly often unintended) changes in behavior when the users adds a second column
    (filtering no longer can trigger at all).
    The default will be changed to [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all] in future versions.
    To silence the warning, set ``filter_func`` explicitly.
    ``max_relative_violations`` indicates what fraction of rows of the given table
    may have values not included in the reference set of unique values. Please note
    that ``UniquesSubset`` and ``UniquesSuperset`` are not symmetrical in this regard.

    By default, the number of occurrences affects the computed fraction of violations.
    To disable this weighting, set ``compare_distinct=True``.
    This argument does not have an effect on the test results for other [`Uniques`][datajudge.constraints.uniques.Uniques] constraints,
    or if ``max_relative_violations`` is 0.

    See [`Uniques`][datajudge.constraints.uniques.Uniques] for further details on ``map_func``, ``reduce_func``,
    and ``output_processors``.
    """
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        uniques_constraints.UniquesSubset(
            ref,
            ref2=ref2,
            max_relative_violations=max_relative_violations,
            compare_distinct=compare_distinct,
            filter_func=filter_func,
            map_func=map_func,
            reduce_func=reduce_func,
            output_processors=output_processors,
            name=name,
            cache_size=cache_size,
        )
    )

add_uniques_superset_constraint

add_uniques_superset_constraint(columns1: list[str], columns2: list[str], max_relative_violations: float = 0, filter_func: Callable[[list[_T]], list[_T]] | None = None, map_func: Callable[[_T], _T] | None = None, reduce_func: Callable[[Collection], Collection] | None = None, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, cache_size=None) -> None

Check if unique values of columns are contained in the reference data.

The UniquesSuperset constraint asserts that reference set of expected values, derived from the unique values in given columns of the reference DataSource, is contained in given columns of a DataSource.

Null values in the columns columns are ignored. To assert the non-existence of them use the add_null_absence_constraint helper method for WithinRequirement. By default, the null filtering does not trigger if multiple columns are fetched at once. It can be configured in more detail by supplying a custom filter_func function. Some exemplary implementations are available as filternull_element, filternull_never, filternull_element_or_tuple_all, filternull_element_or_tuple_any. Passing None as the argument is equivalent to filternull_element but triggers a warning. The current default of filternull_element Cause (possibly often unintended) changes in behavior when the users adds a second column (filtering no longer can trigger at all). The default will be changed to filternull_element_or_tuple_all in future versions. To silence the warning, set filter_func explicitly..

max_relative_violations indicates what fraction of unique values of the given DataSource are not represented in the reference set of unique values. Please note that UniquesSubset and UniquesSuperset are not symmetrical in this regard.

One use of this constraint is to test for consistency in columns with expected categorical values.

See Uniques for further details on map_func, reduce_func, and output_processors.

Source code in src/datajudge/requirements.py

def add_uniques_superset_constraint(
    self,
    columns1: list[str],
    columns2: list[str],
    max_relative_violations: float = 0,
    filter_func: Callable[[list[_T]], list[_T]] | None = None,
    map_func: Callable[[_T], _T] | None = None,
    reduce_func: Callable[[Collection], Collection] | None = None,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    cache_size=None,
) -> None:
    """Check if unique values of columns are contained in the reference data.

    The ``UniquesSuperset`` constraint asserts that reference set of expected values,
    derived from the unique values in given columns of the reference ``DataSource``,
    is contained in given columns of a ``DataSource``.

    Null values in the columns ``columns`` are ignored. To assert the non-existence of them use
    the [`add_null_absence_constraint`][datajudge.requirements.WithinRequirement.add_null_absence_constraint] helper method
    for ``WithinRequirement``.
    By default, the null filtering does not trigger if multiple columns are fetched at once.
    It can be configured in more detail by supplying a custom ``filter_func`` function.
    Some exemplary implementations are available as [`filternull_element`][datajudge.utils.filternull_element],
    [`filternull_never`][datajudge.utils.filternull_never], [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all],
    [`filternull_element_or_tuple_any`][datajudge.utils.filternull_element_or_tuple_any].
    Passing ``None`` as the argument is equivalent to [`filternull_element`][datajudge.utils.filternull_element] but triggers a warning.
    The current default of [`filternull_element`][datajudge.utils.filternull_element]
    Cause (possibly often unintended) changes in behavior when the users adds a second column
    (filtering no longer can trigger at all).
    The default will be changed to [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all] in future versions.
    To silence the warning, set ``filter_func`` explicitly..

    ``max_relative_violations`` indicates what fraction of unique values of the given
    ``DataSource`` are not represented in the reference set of unique values. Please
    note that ``UniquesSubset`` and ``UniquesSuperset`` are not symmetrical in this regard.

    One use of this constraint is to test for consistency in columns with expected
    categorical values.

    See [`Uniques`][datajudge.constraints.uniques.Uniques] for further details on ``map_func``, ``reduce_func``,
    and ``output_processors``.
    """
    ref = DataReference(self._data_source, columns1, condition1)
    ref2 = DataReference(self._data_source2, columns2, condition2)
    self._constraints.append(
        uniques_constraints.UniquesSuperset(
            ref,
            ref2=ref2,
            max_relative_violations=max_relative_violations,
            filter_func=filter_func,
            map_func=map_func,
            reduce_func=reduce_func,
            output_processors=output_processors,
            name=name,
            cache_size=cache_size,
        )
    )

add_varchar_max_length_constraint

add_varchar_max_length_constraint(column1: str, column2: str, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_varchar_max_length_constraint(
    self,
    column1: str,
    column2: str,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        varchar_constraints.VarCharMaxLength(
            ref, ref2=ref2, name=name, cache_size=cache_size
        )
    )

add_varchar_min_length_constraint

add_varchar_min_length_constraint(column1: str, column2: str, condition1: Condition | None = None, condition2: Condition | None = None, name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_varchar_min_length_constraint(
    self,
    column1: str,
    column2: str,
    condition1: Condition | None = None,
    condition2: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    ref = DataReference(self._data_source, [column1], condition1)
    ref2 = DataReference(self._data_source2, [column2], condition2)
    self._constraints.append(
        varchar_constraints.VarCharMinLength(
            ref, ref2=ref2, name=name, cache_size=cache_size
        )
    )

from_expressions classmethod

from_expressions(expression1, expression2, name1: str, name2: str, date_column: str | None = None, date_column2: str | None = None)

Create a BetweenTableRequirement based on sqlalchemy expressions.

Any sqlalchemy object implementing the alias method can be passed as an argument for the expression1 and expression2 parameters. This could, e.g. be a sqlalchemy.Table object or the result of a sqlalchemy.select invocation.

name1 and name2 will be used to represent the expressions in error messages, respectively.

Source code in src/datajudge/requirements.py

@classmethod
def from_expressions(
    cls,
    expression1,
    expression2,
    name1: str,
    name2: str,
    date_column: str | None = None,
    date_column2: str | None = None,
):
    """Create a ``BetweenTableRequirement`` based on sqlalchemy expressions.

    Any sqlalchemy object implementing the ``alias`` method can be passed as an
    argument for the ``expression1`` and ``expression2`` parameters. This could,
    e.g. be a ``sqlalchemy.Table`` object or the result of a ``sqlalchemy.select``
    invocation.

    ``name1`` and ``name2`` will be used to represent the expressions in error messages,
    respectively.
    """
    return cls(
        data_source=ExpressionDataSource(expression1, name1),
        data_source2=ExpressionDataSource(expression2, name2),
        date_column=date_column,
        date_column2=date_column2,
    )

from_raw_queries classmethod

from_raw_queries(query1: str, query2: str, name1: str, name2: str, columns1: list[str] | None = None, columns2: list[str] | None = None, date_column: str | None = None, date_column2: str | None = None)

Create a BetweenRequirement based on raw query strings.

The query1 and query2 parameters can be passed any query string returning rows, e.g. "SELECT * FROM myschema.mytable LIMIT 1337" or "SELECT id, name FROM table1 UNION SELECT id, name FROM table2".

name1 and name2 will be used to represent the queries in error messages, respectively.

If constraints rely on specific columns, these should be provided here via columns1 and columns2 respectively.

Source code in src/datajudge/requirements.py

@classmethod
def from_raw_queries(
    cls,
    query1: str,
    query2: str,
    name1: str,
    name2: str,
    columns1: list[str] | None = None,
    columns2: list[str] | None = None,
    date_column: str | None = None,
    date_column2: str | None = None,
):
    """Create a ``BetweenRequirement`` based on raw query strings.

    The ``query1`` and ``query2`` parameters can be passed any query string returning
    rows, e.g. ``"SELECT * FROM myschema.mytable LIMIT 1337"`` or
    ``"SELECT id, name FROM table1 UNION SELECT id, name FROM table2"``.

    ``name1`` and ``name2`` will be used to represent the queries in error messages,
    respectively.

    If constraints rely on specific columns, these should be provided here via
    ``columns1`` and ``columns2`` respectively.
    """
    return cls(
        data_source=RawQueryDataSource(query1, name1, columns=columns1),
        data_source2=RawQueryDataSource(query2, name2, columns=columns2),
        date_column=date_column,
        date_column2=date_column2,
    )

from_tables classmethod

from_tables(db_name1: str, schema_name1: str, table_name1: str, db_name2: str, schema_name2: str, table_name2: str, date_column: str | None = None, date_column2: str | None = None)

Create a BetweenRequirement based on a table.

Source code in src/datajudge/requirements.py

@classmethod
def from_tables(
    cls,
    db_name1: str,
    schema_name1: str,
    table_name1: str,
    db_name2: str,
    schema_name2: str,
    table_name2: str,
    date_column: str | None = None,
    date_column2: str | None = None,
):
    """Create a ``BetweenRequirement`` based on a table."""
    return cls(
        data_source=TableDataSource(
            db_name=db_name1,
            schema_name=schema_name1,
            table_name=table_name1,
        ),
        data_source2=TableDataSource(
            db_name=db_name2,
            schema_name=schema_name2,
            table_name=table_name2,
        ),
        date_column=date_column,
        date_column2=date_column2,
    )

Requirement

Requirement()

Bases: ABC, MutableSequence

Source code in src/datajudge/requirements.py

def __init__(self):
    self._constraints: list[Constraint] = []
    self._data_source: DataSource

insert

insert(index: int, value: Constraint) -> None

Source code in src/datajudge/requirements.py

def insert(self, index: int, value: Constraint) -> None:
    self._constraints.insert(index, value)

test

test(engine) -> list[TestResult]

Source code in src/datajudge/requirements.py

def test(self, engine) -> list[TestResult]:
    return [constraint.test(engine) for constraint in self]

TableQualifier

TableQualifier(db_name: str, schema_name: str, table_name: str)

Source code in src/datajudge/requirements.py

def __init__(self, db_name: str, schema_name: str, table_name: str):
    self._db_name = db_name
    self._schema_name = schema_name
    self._table_name = table_name

get_between_requirement

get_between_requirement(table_qualifier)

Source code in src/datajudge/requirements.py

def get_between_requirement(self, table_qualifier):
    return BetweenRequirement.from_tables(
        db_name1=self._db_name,
        schema_name1=self._schema_name,
        table_name1=self._table_name,
        db_name2=table_qualifier.db_name,
        schema_name2=table_qualifier.schema_name,
        table_name2=table_qualifier.table_name,
    )

get_within_requirement

get_within_requirement()

Source code in src/datajudge/requirements.py

def get_within_requirement(self):
    return WithinRequirement.from_table(
        db_name=self._db_name,
        schema_name=self._schema_name,
        table_name=self._table_name,
    )

WithinRequirement

WithinRequirement(data_source: DataSource)

Bases: Requirement

Source code in src/datajudge/requirements.py

def __init__(self, data_source: DataSource):
    self._data_source = data_source
    super().__init__()

add_categorical_bound_constraint

add_categorical_bound_constraint(columns: list[str], distribution: dict[_T, tuple[float, float]], default_bounds: tuple[float, float] = (0, 0), max_relative_violations: float = 0, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Check if the distribution of unique values in columns falls within the specified minimum and maximum bounds.

The CategoricalBoundConstraint is added to ensure the distribution of unique values in the specified columns of a DataSource falls within the given minimum and maximum bounds defined in the distribution parameter.

PARAMETER	DESCRIPTION
columns	A list of column names from the DataSource to apply the constraint on. TYPE: list[str]
distribution	A dictionary where keys represent unique values and the corresponding tuple values represent the minimum and maximum allowed proportions of the respective unique value in the columns. TYPE: dict[_T, tuple[float, float]]
default_bounds	A tuple specifying the minimum and maximum allowed proportions for all elements not mentioned in the distribution. By default, it's set to (0, 0), which means all elements not present in distribution will cause a constraint failure. TYPE: tuple[float, float] DEFAULT: (0, 0)
max_relative_violations	A tolerance threshold (0 to 1) for the proportion of elements in the data that can violate the bound constraints without triggering the constraint violation. TYPE: float DEFAULT: 0
condition	An optional parameter to specify a Condition object to filter the data before applying the constraint. TYPE: Condition | None DEFAULT: None
name	An optional parameter to provide a custom name for the constraint. TYPE: str | None DEFAULT: None
cache_size	TODO DEFAULT: None

Example:

This method can be used to test for consistency in columns with expected categorical values or ensure that the distribution of values in a column adheres to a certain criterion.

Usage:

requirement = WithinRequirement(data_source)
requirement.add_categorical_bound_constraint(
    columns=['column_name'],
    distribution={'A': (0.2, 0.3), 'B': (0.4, 0.6), 'C': (0.1, 0.2)},
    max_relative_violations=0.05,
    name='custom_name'
)

Source code in src/datajudge/requirements.py

def add_categorical_bound_constraint(
    self,
    columns: list[str],
    distribution: dict[_T, tuple[float, float]],
    default_bounds: tuple[float, float] = (0, 0),
    max_relative_violations: float = 0,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Check if the distribution of unique values in columns falls within the specified minimum and maximum bounds.

    The ``CategoricalBoundConstraint`` is added to ensure the distribution of unique values
    in the specified columns of a ``DataSource`` falls within the given minimum and maximum
    bounds defined in the ``distribution`` parameter.

    Parameters
    ----------
    columns:
        A list of column names from the `DataSource` to apply the constraint on.
    distribution:
        A dictionary where keys represent unique values and the corresponding
        tuple values represent the minimum and maximum allowed proportions of the respective
        unique value in the columns.
    default_bounds:
        A tuple specifying the minimum and maximum allowed proportions for all
        elements not mentioned in the distribution. By default, it's set to (0, 0), which means
        all elements not present in `distribution` will cause a constraint failure.
    max_relative_violations:
        A tolerance threshold (0 to 1) for the proportion of elements in the data that can violate the
        bound constraints without triggering the constraint violation.
    condition:
        An optional parameter to specify a `Condition` object to filter the data
        before applying the constraint.
    name:
        An optional parameter to provide a custom name for the constraint.
    cache_size:
        TODO

    Example:
    -------
    This method can be used to test for consistency in columns with expected categorical
    values or ensure that the distribution of values in a column adheres to a certain
    criterion.

    Usage:

    ```
    requirement = WithinRequirement(data_source)
    requirement.add_categorical_bound_constraint(
        columns=['column_name'],
        distribution={'A': (0.2, 0.3), 'B': (0.4, 0.6), 'C': (0.1, 0.2)},
        max_relative_violations=0.05,
        name='custom_name'
    )
    ```
    """
    ref = DataReference(self._data_source, columns, condition)
    self._constraints.append(
        uniques_constraints.CategoricalBoundConstraint(
            ref,
            distribution=distribution,
            default_bounds=default_bounds,
            max_relative_violations=max_relative_violations,
            name=name,
            cache_size=cache_size,
        )
    )

add_column_existence_constraint

add_column_existence_constraint(columns: list[str], name: str | None = None, cache_size=None) -> None

Source code in src/datajudge/requirements.py

def add_column_existence_constraint(
    self, columns: list[str], name: str | None = None, cache_size=None
) -> None:
    # Note that columns are not meant to be part of the reference.
    ref = DataReference(self._data_source)
    self._constraints.append(
        column_constraints.ColumnExistence(ref, columns, cache_size=cache_size)
    )

add_column_type_constraint

add_column_type_constraint(column: str, column_type: str | TypeEngine, name: str | None = None, cache_size=None) -> None

Check if a column type matches the expected column_type.

The column_type can be provided as a string (backend-specific type name), a backend-specific SQLAlchemy type, or a SQLAlchemy's generic type.

If SQLAlchemy's generic types are used, the check is performed using isinstance, which means that the actual type can also be a subclass of the target type. For more information on SQLAlchemy's generic types, see https://docs.sqlalchemy.org/en/20/core/type_basics.html

PARAMETER	DESCRIPTION
column	The name of the column to which the constraint will be applied. TYPE: str
column_type	The expected type of the column. This can be a string, a backend-specific SQLAlchemy type, or a generic SQLAlchemy type. TYPE: str | TypeEngine
name	An optional name for the constraint. If not provided, a name will be generated automatically. TYPE: str | None DEFAULT: None

Source code in src/datajudge/requirements.py

def add_column_type_constraint(
    self,
    column: str,
    column_type: str | sa.types.TypeEngine,
    name: str | None = None,
    cache_size=None,
) -> None:
    """
    Check if a column type matches the expected column_type.

    The ``column_type`` can be provided as a string (backend-specific type name), a backend-specific SQLAlchemy type, or a SQLAlchemy's generic type.

    If SQLAlchemy's generic types are used, the check is performed using ``isinstance``, which means that the actual type can also be a subclass of the target type.
    For more information on SQLAlchemy's generic types, see https://docs.sqlalchemy.org/en/20/core/type_basics.html

    Parameters
    ----------
    column : str
        The name of the column to which the constraint will be applied.

    column_type : str | sa.types.TypeEngine
        The expected type of the column. This can be a string, a backend-specific SQLAlchemy type, or a generic SQLAlchemy type.

    name : str | None
        An optional name for the constraint. If not provided, a name will be generated automatically.
    """
    ref = DataReference(self._data_source, [column])
    self._constraints.append(
        column_constraints.ColumnType(
            ref,
            column_type=column_type,
            name=name,
            cache_size=cache_size,
        )
    )

add_date_between_constraint

add_date_between_constraint(column: str, lower_bound: str, upper_bound: str, min_fraction: float, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Use string format: lower_bound="'20121230'".

Source code in src/datajudge/requirements.py

def add_date_between_constraint(
    self,
    column: str,
    lower_bound: str,
    upper_bound: str,
    min_fraction: float,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Use string format: ``lower_bound="'20121230'"``."""
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        date_constraints.DateBetween(
            ref,
            min_fraction,
            lower_bound,
            upper_bound,
            cache_size=cache_size,
        )
    )

add_date_max_constraint

add_date_max_constraint(column: str, max_value: str, use_upper_bound_reference: bool = True, column_type: str = 'date', condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Ensure all dates to be superior than max_value.

Use string format: max_value="'20121230'"

For more information on column_type values, see add_column_type_constraint.

If use_upper_bound_reference is True, the maximum date in column has to be smaller or equal to max_value. Otherwise the maximum date in column has to be greater or equal to max_value.

Source code in src/datajudge/requirements.py

def add_date_max_constraint(
    self,
    column: str,
    max_value: str,
    use_upper_bound_reference: bool = True,
    column_type: str = "date",
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Ensure all dates to be superior than ``max_value``.

    Use string format: ``max_value="'20121230'"``

    For more information on ``column_type`` values, see [`add_column_type_constraint`][datajudge.requirements.WithinRequirement.add_column_type_constraint].

    If ``use_upper_bound_reference`` is ``True``, the maximum date in ``column`` has to be smaller or
    equal to ``max_value``. Otherwise the maximum date in ``column`` has to be greater or equal
    to ``max_value``.
    """
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        date_constraints.DateMax(
            ref,
            max_value=max_value,
            use_upper_bound_reference=use_upper_bound_reference,
            column_type=column_type,
            name=name,
            cache_size=cache_size,
        )
    )

add_date_min_constraint

add_date_min_constraint(column: str, min_value: str, use_lower_bound_reference: bool = True, column_type: str = 'date', condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Ensure all dates to be superior than min_value.

Use string format: min_value="'20121230'".

For more information on column_type values, see add_column_type_constraint.

If use_lower_bound_reference, the min of the first table has to be greater or equal to min_value. If not use_upper_bound_reference, the min of the first table has to be smaller or equal to min_value.

Source code in src/datajudge/requirements.py

def add_date_min_constraint(
    self,
    column: str,
    min_value: str,
    use_lower_bound_reference: bool = True,
    column_type: str = "date",
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Ensure all dates to be superior than ``min_value``.

    Use string format: ``min_value="'20121230'"``.

    For more information on ``column_type`` values, see ``add_column_type_constraint``.

    If ``use_lower_bound_reference``, the min of the first table has to be
    greater or equal to ``min_value``.
    If not ``use_upper_bound_reference``, the min of the first table has to
    be smaller or equal to ``min_value``.
    """
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        date_constraints.DateMin(
            ref,
            min_value=min_value,
            use_lower_bound_reference=use_lower_bound_reference,
            column_type=column_type,
            name=name,
            cache_size=cache_size,
        )
    )

add_date_no_gap_constraint

add_date_no_gap_constraint(start_column: str, end_column: str, key_columns: list[str] | None = None, end_included: bool = True, max_relative_n_violations: float = 0, condition: Condition | None = None, name: str | None = None, cache_size=None)

Express that date range rows have no gap in-between them.

The table under inspection must consist of at least one but up to many key columns, identifying an entity. Additionally, a start_column and an end_column, indicating start and end dates, should be provided.

Neither of those columns should contain NULL values. Also, it should hold that for a given row, the value of end_column is strictly greater than the value of start_column.

Note that the value of start_column is expected to be included in each date range. By default, the value of end_column is expected to be included as well - this can however be changed by setting end_included to False.

A 'key' is a fixed set of values in key_columns and represents an entity of interest. A priori, a key is not a primary key, i.e., a key can have and often has several rows. Thereby, a key will often come with several date ranges.

If key_columns is None or [], all columns of the table will be considered as composing the key.

In order to express a tolerance for some violations of this gap property, use the max_relative_n_violations parameter. The latter expresses for what fraction of all key_values, at least one gap may exist.

For illustrative examples of this constraint, please refer to its test cases.

Source code in src/datajudge/requirements.py

def add_date_no_gap_constraint(
    self,
    start_column: str,
    end_column: str,
    key_columns: list[str] | None = None,
    end_included: bool = True,
    max_relative_n_violations: float = 0,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    """Express that date range rows have no gap in-between them.

    The table under inspection must consist of at least one but up to many key columns,
    identifying an entity. Additionally, a ``start_column`` and an ``end_column``,
    indicating start and end dates, should be provided.

    Neither of those columns should contain ``NULL`` values. Also, it should hold that
    for a given row, the value of ``end_column`` is strictly greater than the value of
    ``start_column``.

    Note that the value of ``start_column`` is expected to be included in each date range.
    By default, the value of ``end_column`` is expected to be included as well - this can
    however be changed by setting ``end_included`` to ``False``.

    A 'key' is a fixed set of values in ``key_columns`` and represents an entity of
    interest. A priori, a key is not a primary key, i.e., a key can have and often has
    several rows. Thereby, a key will often come with several date ranges.

    If ``key_columns`` is ``None`` or ``[]``, all columns of the table will be
    considered as composing the key.

    In order to express a tolerance for some violations of this gap property, use the
    ``max_relative_n_violations`` parameter. The latter expresses for what fraction
    of all key_values, at least one gap may exist.

    For illustrative examples of this constraint, please refer to its test cases.
    """
    relevant_columns = (
        ([start_column, end_column] + key_columns) if key_columns else []
    )
    ref = DataReference(self._data_source, relevant_columns, condition)
    self._constraints.append(
        date_constraints.DateNoGap(
            ref,
            key_columns=key_columns,
            start_columns=[start_column],
            end_columns=[end_column],
            max_relative_n_violations=max_relative_n_violations,
            legitimate_gap_size=1 if end_included else 0,
            name=name,
            cache_size=cache_size,
        )
    )

add_date_no_overlap_2d_constraint

add_date_no_overlap_2d_constraint(start_column1: str, end_column1: str, start_column2: str, end_column2: str, key_columns: list[str] | None = None, end_included: bool = True, max_relative_n_violations: float = 0, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Express that several date range rows do not overlap in two date dimensions.

The table under inspection must consist of at least one but up to many key columns, identifying an entity. Per date dimension, a start column (start_column1, start_column2) and end (end_column1, end_column2) column should be provided in order to define date ranges.

Date ranges in different date dimensions are expected to represent different kinds of dates. For example, let's say that a row in a table indicates an averag temperature forecast. start_column1 and end_column1 could the date span that was forecasted, e.g. the weather from next Saturday to next Sunday. end_column1 and end_column2 might indicate the timespan when this forceast was used, e.g. from the previous Monday to Wednesday.

Neither of those columns should contain NULL values. Also, the value of end_column_k should be strictly greater than the value of start_column_k.

Note that the values of start_column1 and start_column2 are expected to be included in each date range. By default, the values of end_column1 and end_column2 are expected to be included as well - this can however be changed by setting end_included to False.

A 'key' is a fixed set of values in key_columns and represents an entity of interest. A priori, a key is not a primary key, i.e., a key can have and often has several rows. Thereby, a key will often come with several date ranges.

Often, you might want the date ranges for a given key not to overlap.

If key_columns is None or [], all columns of the table will be considered as composing the key.

In order to express a tolerance for some violations of this non-overlapping property, use the max_relative_n_violations parameter. The latter expresses for what fraction of all key_values, at least one overlap may exist.

For illustrative examples of this constraint, please refer to its test cases.

Source code in src/datajudge/requirements.py

def add_date_no_overlap_2d_constraint(
    self,
    start_column1: str,
    end_column1: str,
    start_column2: str,
    end_column2: str,
    key_columns: list[str] | None = None,
    end_included: bool = True,
    max_relative_n_violations: float = 0,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Express that several date range rows do not overlap in two date dimensions.

    The table under inspection must consist of at least one but up to many key columns,
    identifying an entity. Per date dimension, a start column (``start_column1``, ``start_column2``)
    and end (``end_column1``, ``end_column2``) column should be provided in order to define
    date ranges.

    Date ranges in different date dimensions are expected to represent different kinds
    of dates. For example, let's say that a row in a table indicates an averag temperature
    forecast. ``start_column1`` and ``end_column1`` could the date span that was forecasted,
    e.g. the weather from next Saturday to next Sunday. ``end_column1`` and ``end_column2``
    might indicate the timespan when this forceast was used, e.g. from the
    previous Monday to Wednesday.

    Neither of those columns should contain ``NULL`` values. Also, the value of ``end_column_k``
    should be strictly greater than the value of ``start_column_k``.

    Note that the values of ``start_column1`` and ``start_column2`` are expected to be
    included in each date range. By default, the values of ``end_column1`` and
    ``end_column2`` are expected to be included as well - this can however be changed
    by setting ``end_included`` to ``False``.

    A 'key' is a fixed set of values in key_columns and represents an entity of
    interest. A priori, a key is not a primary key, i.e., a key can have and often has
    several rows. Thereby, a key will often come with several date ranges.

    Often, you might want the date ranges for a given key not to overlap.

    If key_columns is ``None`` or ``[]``, all columns of the table will be considered as
    composing the key.

    In order to express a tolerance for some violations of this non-overlapping property,
    use the ``max_relative_n_violations`` parameter. The latter expresses for what fraction
    of all key_values, at least one overlap may exist.

    For illustrative examples of this constraint, please refer to its test cases.
    """
    relevant_columns = (
        [start_column1, end_column1, start_column2, end_column2] + key_columns
        if key_columns
        else []
    )
    ref = DataReference(
        self._data_source,
        relevant_columns,
        condition,
    )
    self._constraints.append(
        date_constraints.DateNoOverlap2d(
            ref,
            key_columns=key_columns,
            start_columns=[start_column1, start_column2],
            end_columns=[end_column1, end_column2],
            end_included=end_included,
            max_relative_n_violations=max_relative_n_violations,
            name=name,
            cache_size=cache_size,
        )
    )

add_date_no_overlap_constraint

add_date_no_overlap_constraint(start_column: str, end_column: str, key_columns: list[str] | None = None, end_included: bool = True, max_relative_n_violations: float = 0, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Constraint expressing that several date range rows may not overlap.

The DataSource under inspection must consist of at least one but up to many key_columns, identifying an entity, a start_column and an end_column.

For a given row in this DataSource, start_column and end_column indicate a date range. Neither of those columns should contain NULL values. Also, it should hold that for a given row, the value of end_column is strictly greater than the value of start_column.

Note that the value of start_column is expected to be included in each date range. By default, the value of end_column is expected to be included as well - this can however be changed by setting end_included to False.

A 'key' is a fixed set of values in key_columns and represents an entity of interest. A priori, a key is not a primary key, i.e., a key can have and often has several rows. Thereby, a key will often come with several date ranges.

Often, you might want the date ranges for a given key not to overlap.

If key_columns is None or [], all columns of the table will be considered as composing the key.

In order to express a tolerance for some violations of this non-overlapping property, use the max_relative_n_violations parameter. The latter expresses for what fraction of all key values, at least one overlap may exist.

For illustrative examples of this constraint, please refer to its test cases.

Source code in src/datajudge/requirements.py

def add_date_no_overlap_constraint(
    self,
    start_column: str,
    end_column: str,
    key_columns: list[str] | None = None,
    end_included: bool = True,
    max_relative_n_violations: float = 0,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Constraint expressing that several date range rows may not overlap.

    The [`DataSource`][datajudge.DataSource] under inspection must consist of at least one but up
    to many ``key_columns``, identifying an entity, a ``start_column`` and an
    ``end_column``.

    For a given row in this [`DataSource`][datajudge.DataSource], ``start_column`` and ``end_column`` indicate a
    date range. Neither of those columns should contain NULL values. Also, it
    should hold that for a given row, the value of ``end_column`` is strictly greater
    than the value of ``start_column``.

    Note that the value of ``start_column`` is expected to be included in each date
    range. By default, the value of ``end_column`` is expected to be included as well -
    this can however be changed by setting ``end_included`` to ``False``.

    A 'key' is a fixed set of values in ``key_columns`` and represents an entity of
    interest. A priori, a key is not a primary key, i.e., a key can have and often
    has several rows. Thereby, a key will often come with several date ranges.

    Often, you might want the date ranges for a given key not to overlap.

    If ``key_columns`` is ``None`` or ``[]``, all columns of the table will be considered
    as composing the key.

    In order to express a tolerance for some violations of this non-overlapping
    property, use the ``max_relative_n_violations`` parameter. The latter expresses for
    what fraction of all key values, at least one overlap may exist.

    For illustrative examples of this constraint, please refer to its test cases.
    """
    relevant_columns = [start_column, end_column] + (
        key_columns if key_columns else []
    )
    ref = DataReference(self._data_source, relevant_columns, condition)
    self._constraints.append(
        date_constraints.DateNoOverlap(
            ref,
            key_columns=key_columns,
            start_columns=[start_column],
            end_columns=[end_column],
            end_included=end_included,
            max_relative_n_violations=max_relative_n_violations,
            name=name,
            cache_size=cache_size,
        )
    )

add_functional_dependency_constraint

add_functional_dependency_constraint(key_columns: list[str], value_columns: list[str], condition: Condition | None = None, name: str | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, cache_size=None)

Expresses a functional dependency, a constraint where the value_columns are uniquely determined by the key_columns.

This means that for each unique combination of values in the key_columns, there is exactly one corresponding combination of values in the value_columns.

The add_unique_constraint constraint is a special case of this constraint, where the key_columns are a primary key, and all other columns are included value_columns. This constraint allows for a more general definition of functional dependencies, where the key_columns are not necessarily a primary key.

An additional configuration option (for details see the analogous parameter in for Uniques-constraints) on how the output is sorted and how many counterexamples are shown is available as output_processors.

An additional configuration option (for details see the analogous parameter in for Uniques-constraints) on how the output is sorted and how many counterexamples are shown is available as output_processors.

For more information on functional dependencies, see https://en.wikipedia.org/wiki/Functional_dependency.

Source code in src/datajudge/requirements.py

def add_functional_dependency_constraint(
    self,
    key_columns: list[str],
    value_columns: list[str],
    condition: Condition | None = None,
    name: str | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    cache_size=None,
):
    """Expresses a functional dependency, a constraint where the `value_columns` are uniquely determined by the `key_columns`.

    This means that for each unique combination of values in the `key_columns`, there is exactly one corresponding combination of values in the `value_columns`.

    The ``add_unique_constraint`` constraint is a special case of this constraint, where the ``key_columns`` are a primary key,
    and all other columns are included ``value_columns``.
    This constraint allows for a more general definition of functional dependencies, where the ``key_columns`` are not necessarily a primary key.

    An additional configuration option (for details see the analogous parameter in for ``Uniques``-constraints)
    on how the output is sorted and how many counterexamples are shown is available as ``output_processors``.

    An additional configuration option (for details see the analogous parameter in for ``Uniques``-constraints)
    on how the output is sorted and how many counterexamples are shown is available as ``output_processors``.

    For more information on functional dependencies, see https://en.wikipedia.org/wiki/Functional_dependency.
    """
    relevant_columns = key_columns + value_columns
    ref = DataReference(self._data_source, relevant_columns, condition)
    self._constraints.append(
        miscs_constraints.FunctionalDependency(
            ref,
            key_columns=key_columns,
            output_processors=output_processors,
            name=name,
            cache_size=cache_size,
        )
    )

add_groupby_aggregation_constraint

add_groupby_aggregation_constraint(columns: Sequence[str], aggregation_column: str, start_value: int, tolerance: float = 0, condition: Condition | None = None, name: str | None = None, cache_size=None)

Check whether array aggregate corresponds to an integer range.

The DataSource is grouped by columns. Sql's array_agg function is then applied to the aggregate_column.

Since we expect aggregate_column to be a numeric column, this leads to a multiset of aggregated values. These values should correspond to the integers ranging from start_value to the cardinality of the multiset.

In order to allow for slight deviations from this pattern, tolerance expresses the fraction of all grouped-by rows, which may be incomplete ranges.

Source code in src/datajudge/requirements.py

def add_groupby_aggregation_constraint(
    self,
    columns: Sequence[str],
    aggregation_column: str,
    start_value: int,
    tolerance: float = 0,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    """Check whether array aggregate corresponds to an integer range.

    The ``DataSource`` is grouped by ``columns``. Sql's ``array_agg`` function is then
    applied to the ``aggregate_column``.

    Since we expect ``aggregate_column`` to be a numeric column, this leads to
    a multiset of aggregated values. These values should correspond to the integers
    ranging from ``start_value`` to the cardinality of the multiset.

    In order to allow for slight deviations from this pattern, ``tolerance`` expresses
    the fraction of all grouped-by rows, which may be incomplete ranges.
    """
    ref = DataReference(self._data_source, list(columns), condition)
    self._constraints.append(
        groupby_constraints.AggregateNumericRangeEquality(
            ref,
            aggregation_column=aggregation_column,
            tolerance=tolerance,
            start_value=start_value,
            name=name,
            cache_size=cache_size,
        )
    )

add_max_null_fraction_constraint

add_max_null_fraction_constraint(column: str, max_null_fraction: float, condition: Condition | None = None, name: str | None = None, cache_size=None)

Assert that column has less than a certain fraction of NULL values.

max_null_fraction is expected to lie within [0, 1].

Source code in src/datajudge/requirements.py

def add_max_null_fraction_constraint(
    self,
    column: str,
    max_null_fraction: float,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    """Assert that ``column`` has less than a certain fraction of ``NULL`` values.

    ``max_null_fraction`` is expected to lie within [0, 1].
    """
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        miscs_constraints.MaxNullFraction(
            ref,
            max_null_fraction=max_null_fraction,
            name=name,
            cache_size=cache_size,
        )
    )

add_n_rows_equality_constraint

add_n_rows_equality_constraint(n_rows: int, condition: Condition | None = None, name: str | None = None, cache_size=None)

Source code in src/datajudge/requirements.py

def add_n_rows_equality_constraint(
    self,
    n_rows: int,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    ref = DataReference(self._data_source, None, condition)
    self._constraints.append(
        nrows_constraints.NRowsEquality(
            ref, n_rows=n_rows, name=name, cache_size=cache_size
        )
    )

add_n_rows_max_constraint

add_n_rows_max_constraint(n_rows_max: int, condition: Condition | None = None, name: str | None = None, cache_size=None)

Source code in src/datajudge/requirements.py

def add_n_rows_max_constraint(
    self,
    n_rows_max: int,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    ref = DataReference(self._data_source, None, condition)
    self._constraints.append(
        nrows_constraints.NRowsMax(
            ref, n_rows=n_rows_max, name=name, cache_size=cache_size
        )
    )

add_n_rows_min_constraint

add_n_rows_min_constraint(n_rows_min: int, condition: Condition | None = None, name: str | None = None, cache_size=None)

Source code in src/datajudge/requirements.py

def add_n_rows_min_constraint(
    self,
    n_rows_min: int,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    ref = DataReference(self._data_source, None, condition)
    self._constraints.append(
        nrows_constraints.NRowsMin(
            ref, n_rows=n_rows_min, name=name, cache_size=cache_size
        )
    )

add_n_uniques_equality_constraint

add_n_uniques_equality_constraint(columns: list[str] | None, n_uniques: int, condition: Condition | None = None, name: str | None = None, cache_size=None)

Source code in src/datajudge/requirements.py

def add_n_uniques_equality_constraint(
    self,
    columns: list[str] | None,
    n_uniques: int,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    ref = DataReference(self._data_source, columns, condition)
    self._constraints.append(
        uniques_constraints.NUniquesEquality(
            ref, n_uniques=n_uniques, name=name, cache_size=cache_size
        )
    )

add_null_absence_constraint

add_null_absence_constraint(column: str, condition: Condition | None = None, name: str | None = None, cache_size=None)

Source code in src/datajudge/requirements.py

def add_null_absence_constraint(
    self,
    column: str,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        miscs_constraints.MaxNullFraction(
            ref, max_null_fraction=0, name=name, cache_size=cache_size
        )
    )

add_numeric_between_constraint

add_numeric_between_constraint(column: str, lower_bound: float, upper_bound: float, min_fraction: float, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the column's values lie between lower_bound and upper_bound.

Note that both bounds are inclusive.

Unless specified otherwise via the usage of a condition, NULL values will be considered in the denominator of min_fraction. NULL values will never be considered to lie in the interval [lower_bound, upper_bound].

Source code in src/datajudge/requirements.py

def add_numeric_between_constraint(
    self,
    column: str,
    lower_bound: float,
    upper_bound: float,
    min_fraction: float,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Assert that the column's values lie between ``lower_bound`` and ``upper_bound``.

    Note that both bounds are inclusive.

    Unless specified otherwise via the usage of a ``condition``, ``NULL`` values will
    be considered in the denominator of ``min_fraction``. ``NULL`` values will never be
    considered to lie in the interval [``lower_bound``, ``upper_bound``].
    """
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        numeric_constraints.NumericBetween(
            ref,
            min_fraction,
            lower_bound,
            upper_bound,
            name=name,
            cache_size=cache_size,
        )
    )

add_numeric_max_constraint

add_numeric_max_constraint(column: str, max_value: float, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

All values in column are less or equal max_value.

Source code in src/datajudge/requirements.py

def add_numeric_max_constraint(
    self,
    column: str,
    max_value: float,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """All values in ``column`` are less or equal ``max_value``."""
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        numeric_constraints.NumericMax(
            ref, max_value=max_value, name=name, cache_size=cache_size
        )
    )

add_numeric_mean_constraint

add_numeric_mean_constraint(column: str, mean_value: float, max_absolute_deviation: float, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert the mean of the column column deviates at most max_deviation from mean_value.

Source code in src/datajudge/requirements.py

def add_numeric_mean_constraint(
    self,
    column: str,
    mean_value: float,
    max_absolute_deviation: float,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Assert the mean of the column ``column`` deviates at most ``max_deviation`` from ``mean_value``."""
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        numeric_constraints.NumericMean(
            ref,
            max_absolute_deviation,
            mean_value=mean_value,
            name=name,
            cache_size=cache_size,
        )
    )

add_numeric_min_constraint

add_numeric_min_constraint(column: str, min_value: float, condition: Condition | None = None, cache_size=None) -> None

All values in column are greater or equal min_value.

Source code in src/datajudge/requirements.py

def add_numeric_min_constraint(
    self,
    column: str,
    min_value: float,
    condition: Condition | None = None,
    cache_size=None,
) -> None:
    """All values in ``column`` are greater or equal ``min_value``."""
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        numeric_constraints.NumericMin(
            ref, min_value=min_value, cache_size=cache_size
        )
    )

add_numeric_no_gap_constraint

add_numeric_no_gap_constraint(start_column: str, end_column: str, key_columns: list[str] | None = None, legitimate_gap_size: float = 0, max_relative_n_violations: float = 0, condition: Condition | None = None, name: str | None = None, cache_size=None)

Express that numeric interval rows have no gaps larger than some max value in-between them.

The table under inspection must consist of at least one but up to many key columns, identifying an entity. Additionally, a start_column and an end_column, indicating interval start and end values, should be provided.

Neither of those columns should contain NULL values. Also, it should hold that for a given row, the value of end_column is strictly greater than the value of start_column.

legitimate_gap_size is the maximum tollerated gap size between two intervals.

A 'key' is a fixed set of values in key_columns and represents an entity of interest. A priori, a key is not a primary key, i.e., a key can have and often has several rows. Thereby, a key will often come with several intervals.

If key_columns is None or [], all columns of the table will be considered as composing the key.

In order to express a tolerance for some violations of this gap property, use the max_relative_n_violations parameter. The latter expresses for what fraction of all key_values, at least one gap may exist.

For illustrative examples of this constraint, please refer to its test cases.

Source code in src/datajudge/requirements.py

def add_numeric_no_gap_constraint(
    self,
    start_column: str,
    end_column: str,
    key_columns: list[str] | None = None,
    legitimate_gap_size: float = 0,
    max_relative_n_violations: float = 0,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    """Express that numeric interval rows have no gaps larger than some max value in-between them.

    The table under inspection must consist of at least one but up to many key columns,
    identifying an entity. Additionally, a ``start_column`` and an ``end_column``,
    indicating interval start and end values, should be provided.

    Neither of those columns should contain ``NULL`` values. Also, it should hold that
    for a given row, the value of ``end_column`` is strictly greater than the value of
    ``start_column``.

    ``legitimate_gap_size`` is the maximum tollerated gap size between two intervals.

    A 'key' is a fixed set of values in ``key_columns`` and represents an entity of
    interest. A priori, a key is not a primary key, i.e., a key can have and often has
    several rows. Thereby, a key will often come with several intervals.

    If ``key_columns`` is ``None`` or ``[]``, all columns of the table will be
    considered as composing the key.

    In order to express a tolerance for some violations of this gap property, use the
    ``max_relative_n_violations`` parameter. The latter expresses for what fraction
    of all key_values, at least one gap may exist.

    For illustrative examples of this constraint, please refer to its test cases.
    """
    relevant_columns = (
        ([start_column, end_column] + key_columns) if key_columns else []
    )
    ref = DataReference(self._data_source, relevant_columns, condition)
    self._constraints.append(
        numeric_constraints.NumericNoGap(
            ref,
            key_columns=key_columns,
            start_columns=[start_column],
            end_columns=[end_column],
            legitimate_gap_size=legitimate_gap_size,
            max_relative_n_violations=max_relative_n_violations,
            name=name,
            cache_size=cache_size,
        )
    )

add_numeric_no_overlap_constraint

add_numeric_no_overlap_constraint(start_column: str, end_column: str, key_columns: list[str] | None = None, end_included: bool = True, max_relative_n_violations: float = 0, condition: Condition | None = None, name: str | None = None, cache_size=None)

Constraint expressing that several numeric interval rows may not overlap.

The DataSource under inspection must consist of at least one but up to many key_columns, identifying an entity, a start_column and an end_column.

For a given row in this DataSource, start_column and end_column indicate a numeric interval. Neither of those columns should contain NULL values. Also, it should hold that for a given row, the value of end_column is strictly greater than the value of start_column.

Note that the value of start_column is expected to be included in each interval. By default, the value of end_column is expected to be included as well - this can however be changed by setting end_included to False.

A 'key' is a fixed set of values in key_columns and represents an entity of interest. A priori, a key is not a primary key, i.e., a key can have and often has several rows. Thereby, a key will often come with several intervals.

Often, you might want the intervals for a given key not to overlap.

If key_columns is None or [], all columns of the table will be considered as composing the key.

In order to express a tolerance for some violations of this non-overlapping property, use the max_relative_n_violations parameter. The latter expresses for what fraction of all key values, at least one overlap may exist.

For illustrative examples of this constraint, please refer to its test cases.

Source code in src/datajudge/requirements.py

def add_numeric_no_overlap_constraint(
    self,
    start_column: str,
    end_column: str,
    key_columns: list[str] | None = None,
    end_included: bool = True,
    max_relative_n_violations: float = 0,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    """Constraint expressing that several numeric interval rows may not overlap.

    The ``DataSource`` under inspection must consist of at least one but up
    to many ``key_columns``, identifying an entity, a ``start_column`` and an
    ``end_column``.

    For a given row in this ``DataSource``, ``start_column`` and ``end_column`` indicate a
    numeric interval. Neither of those columns should contain NULL values. Also, it
    should hold that for a given row, the value of ``end_column`` is strictly greater
    than the value of ``start_column``.

    Note that the value of ``start_column`` is expected to be included in each interval.
    By default, the value of ``end_column`` is expected to be included as well -
    this can however be changed by setting ``end_included`` to ``False``.

    A 'key' is a fixed set of values in ``key_columns`` and represents an entity of
    interest. A priori, a key is not a primary key, i.e., a key can have and often
    has several rows. Thereby, a key will often come with several intervals.

    Often, you might want the intervals for a given key not to overlap.

    If ``key_columns`` is ``None`` or ``[]``, all columns of the table will be considered
    as composing the key.

    In order to express a tolerance for some violations of this non-overlapping
    property, use the ``max_relative_n_violations`` parameter. The latter expresses for
    what fraction of all key values, at least one overlap may exist.

    For illustrative examples of this constraint, please refer to its test cases.
    """
    relevant_columns = [start_column, end_column] + (
        key_columns if key_columns else []
    )
    ref = DataReference(self._data_source, relevant_columns, condition)
    self._constraints.append(
        numeric_constraints.NumericNoOverlap(
            ref,
            key_columns=key_columns,
            start_columns=[start_column],
            end_columns=[end_column],
            end_included=end_included,
            max_relative_n_violations=max_relative_n_violations,
            name=name,
            cache_size=cache_size,
        )
    )

add_numeric_percentile_constraint

add_numeric_percentile_constraint(column: str, percentage: float, expected_percentile: float, max_absolute_deviation: float | None = None, max_relative_deviation: float | None = None, condition: Condition | None = None, name: str | None = None, cache_size=None) -> None

Assert that the percentage-th percentile is approximately expected_percentile.

The percentile is defined as the smallest value present in column for which percentage % of the values in column are less or equal. NULL values are ignored.

Hence, if percentage is less than the inverse of the number of non-NULL rows, None is received as the percentage -th percentile.

percentage is expected to be provided in percent. The median, for example, would correspond to percentage=50.

At least one of max_absolute_deviation and max_relative_deviation must be provided.

Source code in src/datajudge/requirements.py

def add_numeric_percentile_constraint(
    self,
    column: str,
    percentage: float,
    expected_percentile: float,
    max_absolute_deviation: float | None = None,
    max_relative_deviation: float | None = None,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Assert that the ``percentage``-th percentile is approximately ``expected_percentile``.

    The percentile is defined as the smallest value present in ``column`` for which
    ``percentage`` % of the values in ``column`` are less or equal. ``NULL`` values
    are ignored.

    Hence, if ``percentage`` is less than the inverse of the number of non-``NULL`` rows,
    ``None`` is received as the ``percentage`` -th percentile.

    ``percentage`` is expected to be provided in percent. The median, for example, would
    correspond to ``percentage=50``.

    At least one of ``max_absolute_deviation`` and ``max_relative_deviation`` must
    be provided.
    """
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        numeric_constraints.NumericPercentile(
            ref,
            percentage=percentage,
            expected_percentile=expected_percentile,
            max_absolute_deviation=max_absolute_deviation,
            max_relative_deviation=max_relative_deviation,
            name=name,
            cache_size=cache_size,
        )
    )

add_primary_key_definition_constraint

add_primary_key_definition_constraint(primary_keys: list[str], name: str | None = None, cache_size=None) -> None

Check that the primary key constraints in the database are exactly equal to the given column names.

Note that this doesn't actually check that the primary key values are unique across the table.

Source code in src/datajudge/requirements.py

def add_primary_key_definition_constraint(
    self,
    primary_keys: list[str],
    name: str | None = None,
    cache_size=None,
) -> None:
    """Check that the primary key constraints in the database are exactly equal to the given column names.

    Note that this doesn't actually check that the primary key values are unique across the table.
    """
    ref = DataReference(self._data_source)
    self._constraints.append(
        miscs_constraints.PrimaryKeyDefinition(
            ref, primary_keys, name=name, cache_size=cache_size
        )
    )

add_uniqueness_constraint

add_uniqueness_constraint(columns: list[str] | None = None, max_duplicate_fraction: float = 0, condition: Condition | None = None, max_absolute_n_duplicates: int = 0, infer_pk_columns: bool = False, name: str | None = None, cache_size=None) -> None

Columns should uniquely identify row.

Given a list of columns columns, validate the condition of a primary key, i.e. uniqueness of tuples in said columns. This constraint has a tolerance for inconsistencies, expressed via max_duplicate_fraction. The latter suggests that the number of uniques from said columns is larger or equal to 1 - max_duplicate_fraction times the number of rows.

If infer_pk_columns is True, columns will be retrieved from the primary keys. If columns is None and infer_pk_column is False, the fallback is validating that all rows in a table are unique.

Source code in src/datajudge/requirements.py

def add_uniqueness_constraint(
    self,
    columns: list[str] | None = None,
    max_duplicate_fraction: float = 0,
    condition: Condition | None = None,
    max_absolute_n_duplicates: int = 0,
    infer_pk_columns: bool = False,
    name: str | None = None,
    cache_size=None,
) -> None:
    """Columns should uniquely identify row.

    Given a list of columns ``columns``, validate the condition of a primary key, i.e.
    uniqueness of tuples in said columns. This constraint has a tolerance
    for inconsistencies, expressed via ``max_duplicate_fraction``. The latter
    suggests that the number of uniques from said columns is larger or equal
    to ``1 - max_duplicate_fraction`` times the number of rows.

    If ``infer_pk_columns`` is ``True``, ``columns`` will be retrieved from the primary keys.
    If ``columns`` is ``None`` and ``infer_pk_column`` is ``False``, the fallback is
    validating that all rows in a table are unique.
    """
    ref = DataReference(self._data_source, columns, condition)
    self._constraints.append(
        miscs_constraints.Uniqueness(
            ref,
            max_duplicate_fraction=max_duplicate_fraction,
            max_absolute_n_duplicates=max_absolute_n_duplicates,
            infer_pk_columns=infer_pk_columns,
            name=name,
            cache_size=cache_size,
        )
    )

add_uniques_equality_constraint

add_uniques_equality_constraint(columns: list[str], uniques: Collection[_T], filter_func: Callable[[list[_T]], list[_T]] | None = None, map_func: Callable[[_T], _T] | None = None, reduce_func: Callable[[Collection], Collection] | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, condition: Condition | None = None, name: str | None = None, cache_size=None)

Check if the data's unique values are equal to a given set of values.

The UniquesEquality constraint asserts if the values contained in a column of a DataSource are strictly the ones of a reference set of expected values, specified via the uniques parameter.

Null values in the columns columns are ignored. To assert the non-existence of them use the add_null_absence_constraint helper method for WithinRequirement.

By default, the null filtering does not trigger if multiple columns are fetched at once. It can be configured in more detail by supplying a custom filter_func function. Some exemplary implementations are available as filternull_element, filternull_never, filternull_element_or_tuple_all, filternull_element_or_tuple_any. Passing None as the argument is equivalent to filternull_element but triggers a warning. The current default of filternull_element Cause (possibly often unintended) changes in behavior when the users adds a second column (filtering no longer can trigger at all). The default will be changed to filternull_element_or_tuple_all in future versions. To silence the warning, set filter_func explicitly.

See the Uniques class for further parameter details on map_func and reduce_func, and output_processors.

Source code in src/datajudge/requirements.py

def add_uniques_equality_constraint(
    self,
    columns: list[str],
    uniques: Collection[_T],
    filter_func: Callable[[list[_T]], list[_T]] | None = None,
    map_func: Callable[[_T], _T] | None = None,
    reduce_func: Callable[[Collection], Collection] | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    """Check if the data's unique values are equal to a given set of values.

    The `UniquesEquality` constraint asserts if the values contained in a column
    of a `DataSource` are strictly the ones of a reference set of expected values,
    specified via the `uniques` parameter.

    Null values in the columns `columns` are ignored. To assert the non-existence of them use
    the [`add_null_absence_constraint`][datajudge.requirements.WithinRequirement.add_null_absence_constraint] helper method
    for `WithinRequirement`.

    By default, the null filtering does not trigger if multiple columns are fetched at once.
    It can be configured in more detail by supplying a custom `filter_func` function.
    Some exemplary implementations are available as [`filternull_element`][datajudge.utils.filternull_element],
    [`filternull_never`][datajudge.utils.filternull_never], [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all],
    [`filternull_element_or_tuple_any`][datajudge.utils.filternull_element_or_tuple_any].
    Passing `None` as the argument is equivalent to [`filternull_element`][datajudge.utils.filternull_element] but triggers a warning.
    The current default of [`filternull_element`][datajudge.utils.filternull_element]
    Cause (possibly often unintended) changes in behavior when the users adds a second column
    (filtering no longer can trigger at all).
    The default will be changed to [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all] in future versions.
    To silence the warning, set `filter_func` explicitly.

    See the `Uniques` class for further parameter details on `map_func` and
    `reduce_func`, and `output_processors`.
    """
    ref = DataReference(self._data_source, columns, condition)
    self._constraints.append(
        uniques_constraints.UniquesEquality(
            ref,
            uniques=uniques,
            filter_func=filter_func,
            map_func=map_func,
            reduce_func=reduce_func,
            output_processors=output_processors,
            name=name,
            cache_size=cache_size,
        )
    )

add_uniques_subset_constraint

add_uniques_subset_constraint(columns: list[str], uniques: Collection[_T], max_relative_violations: float = 0, filter_func: Callable[[list[_T]], list[_T]] | None = None, compare_distinct: bool = False, map_func: Callable[[_T], _T] | None = None, reduce_func: Callable[[Collection], Collection] | None = None, condition: Condition | None = None, name: str | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, cache_size=None)

Check if the data's unique values are contained in a given set of values.

The UniquesSubset constraint asserts if the values contained in a column of a DataSource are part of a reference set of expected values, specified via uniques.

Null values in the columns columns are ignored. To assert the non-existence of them use the add_null_absence_constraint helper method for WithinRequirement. By default, the null filtering does not trigger if multiple columns are fetched at once. It can be configured in more detail by supplying a custom filter_func function. Some exemplary implementations are available as filternull_element, filternull_never, filternull_element_or_tuple_all, filternull_element_or_tuple_any. Passing None as the argument is equivalent to filternull_element but triggers a warning. The current default of filternull_element Cause (possibly often unintended) changes in behavior when the users adds a second column (filtering no longer can trigger at all). The default will be changed to filternull_element_or_tuple_all in future versions. To silence the warning, set filter_func explicitly.

max_relative_violations indicates what fraction of rows of the given table may have values not included in the reference set of unique values. Please note that UniquesSubset and UniquesSuperset are not symmetrical in this regard.

By default, the number of occurrences affects the computed fraction of violations. To disable this weighting, set compare_distinct=True. This argument does not have an effect on the test results for other Uniques constraints, or if max_relative_violations is 0.

See Uniques for further details on map_func, reduce_func, and output_processors.

Source code in src/datajudge/requirements.py

def add_uniques_subset_constraint(
    self,
    columns: list[str],
    uniques: Collection[_T],
    max_relative_violations: float = 0,
    filter_func: Callable[[list[_T]], list[_T]] | None = None,
    compare_distinct: bool = False,
    map_func: Callable[[_T], _T] | None = None,
    reduce_func: Callable[[Collection], Collection] | None = None,
    condition: Condition | None = None,
    name: str | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    cache_size=None,
):
    """Check if the data's unique values are contained in a given set of values.

    The ``UniquesSubset`` constraint asserts if the values contained in a column of
    a ``DataSource`` are part of a reference set of expected values, specified via
    ``uniques``.

    Null values in the columns ``columns`` are ignored. To assert the non-existence of them use
    the [`add_null_absence_constraint`][datajudge.requirements.WithinRequirement.add_null_absence_constraint] helper method
    for ``WithinRequirement``.
    By default, the null filtering does not trigger if multiple columns are fetched at once.
    It can be configured in more detail by supplying a custom ``filter_func`` function.
    Some exemplary implementations are available as [`filternull_element`][datajudge.utils.filternull_element],
    [`filternull_never`][datajudge.utils.filternull_never], [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all],
    [`filternull_element_or_tuple_any`][datajudge.utils.filternull_element_or_tuple_any].
    Passing ``None`` as the argument is equivalent to [`filternull_element`][datajudge.utils.filternull_element] but triggers a warning.
    The current default of [`filternull_element`][datajudge.utils.filternull_element]
    Cause (possibly often unintended) changes in behavior when the users adds a second column
    (filtering no longer can trigger at all).
    The default will be changed to [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all] in future versions.
    To silence the warning, set ``filter_func`` explicitly.


    ``max_relative_violations`` indicates what fraction of rows of the given table
    may have values not included in the reference set of unique values. Please note
    that ``UniquesSubset`` and ``UniquesSuperset`` are not symmetrical in this regard.

    By default, the number of occurrences affects the computed fraction of violations.
    To disable this weighting, set `compare_distinct=True`.
    This argument does not have an effect on the test results for other `Uniques` constraints,
    or if `max_relative_violations` is 0.

    See ``Uniques`` for further details on ``map_func``, ``reduce_func``,
    and ``output_processors``.
    """
    ref = DataReference(self._data_source, columns, condition)
    self._constraints.append(
        uniques_constraints.UniquesSubset(
            ref,
            uniques=uniques,
            max_relative_violations=max_relative_violations,
            filter_func=filter_func,
            compare_distinct=compare_distinct,
            map_func=map_func,
            reduce_func=reduce_func,
            output_processors=output_processors,
            name=name,
            cache_size=cache_size,
        )
    )

add_uniques_superset_constraint

add_uniques_superset_constraint(columns: list[str], uniques: Collection[_T], max_relative_violations: float = 0, filter_func: Callable[[list[_T]], list[_T]] | None = None, map_func: Callable[[_T], _T] | None = None, reduce_func: Callable[[Collection], Collection] | None = None, condition: Condition | None = None, name: str | None = None, output_processors: OutputProcessor | list[OutputProcessor] | None = output_processor_limit, cache_size=None)

Check if unique values of columns are contained in the reference data.

The UniquesSuperset constraint asserts that reference set of expected values, specified via uniques, is contained in given columns of a DataSource.

Null values in the columns columns are ignored. To assert the non-existence of them use the add_null_absence_constraint helper method for WithinRequirement.

By default, the null filtering does not trigger if multiple columns are fetched at once. It can be configured in more detail by supplying a custom filter_func function. Some exemplary implementations are available as filternull_element, filternull_never, filternull_element_or_tuple_all, filternull_element_or_tuple_any. Passing None as the argument is equivalent to filternull_element but triggers a warning. The current default of filternull_element will cause (possibly often unintended) changes in behavior when the user adds a second column (filtering no longer can trigger at all). The default will be changed to filternull_element_or_tuple_all in future versions. To silence the warning, set filter_func explicitly.

max_relative_violations indicates what fraction of unique values of the given DataSource are not represented in the reference set of unique values. Please note that UniquesSubset and UniquesSuperset are not symmetrical in this regard.

One use of this constraint is to test for consistency in columns with expected categorical values.

See Uniques for further details on map_func, reduce_func, and output_processors.

Source code in src/datajudge/requirements.py

def add_uniques_superset_constraint(
    self,
    columns: list[str],
    uniques: Collection[_T],
    max_relative_violations: float = 0,
    filter_func: Callable[[list[_T]], list[_T]] | None = None,
    map_func: Callable[[_T], _T] | None = None,
    reduce_func: Callable[[Collection], Collection] | None = None,
    condition: Condition | None = None,
    name: str | None = None,
    output_processors: OutputProcessor
    | list[OutputProcessor]
    | None = output_processor_limit,
    cache_size=None,
):
    """Check if unique values of columns are contained in the reference data.

    The ``UniquesSuperset`` constraint asserts that reference set of expected values,
    specified via ``uniques``, is contained in given columns of a ``DataSource``.

    Null values in the columns ``columns`` are ignored. To assert the non-existence of them use
    the [`add_null_absence_constraint`][datajudge.requirements.WithinRequirement.add_null_absence_constraint] helper method
    for ``WithinRequirement``.

    By default, the null filtering does not trigger if multiple columns are fetched at once.
    It can be configured in more detail by supplying a custom ``filter_func`` function.
    Some exemplary implementations are available as [`filternull_element`][datajudge.utils.filternull_element],
    [`filternull_never`][datajudge.utils.filternull_never], [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all],
    [`filternull_element_or_tuple_any`][datajudge.utils.filternull_element_or_tuple_any].
    Passing ``None`` as the argument is equivalent to [`filternull_element`][datajudge.utils.filternull_element] but triggers a warning.
    The current default of [`filternull_element`][datajudge.utils.filternull_element]
    will cause (possibly often unintended) changes in behavior when the user adds a second column
    (filtering no longer can trigger at all).
    The default will be changed to [`filternull_element_or_tuple_all`][datajudge.utils.filternull_element_or_tuple_all] in future versions.
    To silence the warning, set ``filter_func`` explicitly.

    ``max_relative_violations`` indicates what fraction of unique values of the given
    ``DataSource`` are not represented in the reference set of unique values. Please
    note that ``UniquesSubset`` and ``UniquesSuperset`` are not symmetrical in this regard.

    One use of this constraint is to test for consistency in columns with expected
    categorical values.

    See ``Uniques`` for further details on ``map_func``, ``reduce_func``,
    and ``output_processors``.
    """
    ref = DataReference(self._data_source, columns, condition)
    self._constraints.append(
        uniques_constraints.UniquesSuperset(
            ref,
            uniques=uniques,
            max_relative_violations=max_relative_violations,
            filter_func=filter_func,
            map_func=map_func,
            reduce_func=reduce_func,
            output_processors=output_processors,
            name=name,
            cache_size=cache_size,
        )
    )

add_varchar_max_length_constraint

add_varchar_max_length_constraint(column: str, max_length: int, condition: Condition | None = None, name: str | None = None, cache_size=None)

Source code in src/datajudge/requirements.py

def add_varchar_max_length_constraint(
    self,
    column: str,
    max_length: int,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        varchar_constraints.VarCharMaxLength(
            ref,
            max_length=max_length,
            name=name,
            cache_size=cache_size,
        )
    )

add_varchar_min_length_constraint

add_varchar_min_length_constraint(column: str, min_length: int, condition: Condition | None = None, name: str | None = None, cache_size=None)

Source code in src/datajudge/requirements.py

def add_varchar_min_length_constraint(
    self,
    column: str,
    min_length: int,
    condition: Condition | None = None,
    name: str | None = None,
    cache_size=None,
):
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        varchar_constraints.VarCharMinLength(
            ref,
            min_length=min_length,
            name=name,
            cache_size=cache_size,
        )
    )

add_varchar_regex_constraint

add_varchar_regex_constraint(column: str, regex: str, condition: Condition | None = None, name: str | None = None, allow_none: bool = False, relative_tolerance: float = 0.0, aggregated: bool = True, n_counterexamples: int = 5, cache_size=None)

Assesses whether the values in a column match a given regular expression pattern.

The option allow_none can be used in cases where the column is defined as nullable and contains null values.

How the tolerance factor is calculated can be controlled with the aggregated flag. When True, the tolerance is calculated using unique values. If not, the tolerance is calculated using all the instances of the data.

n_counterexamples defines how many counterexamples are displayed in an assertion text. If all counterexamples are meant to be shown, provide -1 as an argument.

When using this method, the regex matching will take place in memory. If instead, you would like the matching to take place in database which is typically faster and substantially more memory-saving, please consider using add_varchar_regex_constraint_db.

Source code in src/datajudge/requirements.py

def add_varchar_regex_constraint(
    self,
    column: str,
    regex: str,
    condition: Condition | None = None,
    name: str | None = None,
    allow_none: bool = False,
    relative_tolerance: float = 0.0,
    aggregated: bool = True,
    n_counterexamples: int = 5,
    cache_size=None,
):
    """Assesses whether the values in a column match a given regular expression pattern.

    The option ``allow_none`` can be used in cases where the column is defined as
    nullable and contains null values.

    How the tolerance factor is calculated can be controlled with the ``aggregated``
    flag. When ``True``, the tolerance is calculated using unique values. If not, the
    tolerance is calculated using all the instances of the data.

    ``n_counterexamples`` defines how many counterexamples are displayed in an
    assertion text. If all counterexamples are meant to be shown, provide ``-1`` as
    an argument.

    When using this method, the regex matching will take place in memory. If instead,
    you would like the matching to take place in database which is typically faster and
    substantially more memory-saving, please consider using
    ``add_varchar_regex_constraint_db``.
    """
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        varchar_constraints.VarCharRegex(
            ref,
            regex,
            allow_none=allow_none,
            relative_tolerance=relative_tolerance,
            aggregated=aggregated,
            n_counterexamples=n_counterexamples,
            name=name,
            cache_size=cache_size,
        )
    )

add_varchar_regex_constraint_db

add_varchar_regex_constraint_db(column: str, regex: str, condition: Condition | None = None, name: str | None = None, relative_tolerance: float = 0.0, aggregated: bool = True, n_counterexamples: int = 5, cache_size=None)

Assesses whether the values in a column match a given regular expression pattern.

How the tolerance factor is calculated can be controlled with the aggregated flag. When True, the tolerance is calculated using unique values. If not, the tolerance is calculated using all the instances of the data.

n_counterexamples defines how many counterexamples are displayed in an assertion text. If all counterexamples are meant to be shown, provide -1 as an argument.

When using this method, the regex matching will take place in database, which is only supported for Postgres, Sqllite and Snowflake. Note that for this feature is only for Snowflake when using sqlalchemy-snowflake >= 1.4.0. As an alternative, add_varchar_regex_constraint performs the regex matching in memory. This is typically slower and more expensive in terms of memory but available on all supported database mamangement systems.

Source code in src/datajudge/requirements.py

def add_varchar_regex_constraint_db(
    self,
    column: str,
    regex: str,
    condition: Condition | None = None,
    name: str | None = None,
    relative_tolerance: float = 0.0,
    aggregated: bool = True,
    n_counterexamples: int = 5,
    cache_size=None,
):
    """Assesses whether the values in a column match a given regular expression pattern.

    How the tolerance factor is calculated can be controlled with the ``aggregated``
    flag. When ``True``, the tolerance is calculated using unique values. If not, the
    tolerance is calculated using all the instances of the data.

    ``n_counterexamples`` defines how many counterexamples are displayed in an
    assertion text. If all counterexamples are meant to be shown, provide ``-1`` as
    an argument.

    When using this method, the regex matching will take place in database, which is
    only supported for Postgres, Sqllite and Snowflake. Note that for this
    feature is only for Snowflake when using sqlalchemy-snowflake >= 1.4.0. As an
    alternative, ``add_varchar_regex_constraint`` performs the regex matching in memory.
    This is typically slower and more expensive in terms of memory but available
    on all supported database mamangement systems.
    """
    ref = DataReference(self._data_source, [column], condition)
    self._constraints.append(
        varchar_constraints.VarCharRegexDb(
            ref,
            regex=regex,
            relative_tolerance=relative_tolerance,
            aggregated=aggregated,
            n_counterexamples=n_counterexamples,
            name=name,
            cache_size=cache_size,
        )
    )

from_expression classmethod

from_expression(expression: FromClause, name: str)

Create a WithinRequirement based on a sqlalchemy expression.

Any sqlalchemy object implementing the alias method can be passed as an argument for the expression parameter. This could, e.g. be an sqlalchemy.Table object or the result of a sqlalchemy.select call.

The name will be used to represent this expression in error messages.

Source code in src/datajudge/requirements.py

@classmethod
def from_expression(cls, expression: sa.sql.selectable.FromClause, name: str):
    """Create a ``WithinRequirement`` based on a sqlalchemy expression.

    Any sqlalchemy object implementing the ``alias`` method can be passed as an
    argument for the ``expression`` parameter. This could, e.g. be an
    ``sqlalchemy.Table`` object or the result of a ``sqlalchemy.select`` call.

    The ``name`` will be used to represent this expression in error messages.
    """
    return cls(data_source=ExpressionDataSource(expression, name))

from_raw_query classmethod

from_raw_query(query: str, name: str, columns: list[str] | None = None)

Create a WithinRequirement based on a raw query string.

The query parameter can be passed any query string returning rows, e.g. "SELECT * FROM myschema.mytable LIMIT 1337" or "SELECT id, name FROM table1 UNION SELECT id, name FROM table2".

The name will be used to represent this query in error messages.

If constraints rely on specific columns, these should be provided here via columns, e.g. ["id", "name"].

Source code in src/datajudge/requirements.py

@classmethod
def from_raw_query(cls, query: str, name: str, columns: list[str] | None = None):
    """Create a ``WithinRequirement`` based on a raw query string.

    The ``query`` parameter can be passed any query string returning rows, e.g.
    ``"SELECT * FROM myschema.mytable LIMIT 1337"`` or
    ``"SELECT id, name FROM table1 UNION SELECT id, name FROM table2"``.

    The ``name`` will be used to represent this query in error messages.

    If constraints rely on specific columns, these should be provided here via
    ``columns``, e.g. ``["id", "name"]``.
    """
    return cls(data_source=RawQueryDataSource(query, name, columns=columns))

from_table classmethod

from_table(db_name: str, schema_name: str, table_name: str)

Create a WithinRequirement based on a table.

Source code in src/datajudge/requirements.py

@classmethod
def from_table(cls, db_name: str, schema_name: str, table_name: str):
    """Create a `WithinRequirement` based on a table."""
    return cls(
        data_source=TableDataSource(
            db_name=db_name, schema_name=schema_name, table_name=table_name
        )
    )

utils

OutputProcessor

Bases: Protocol

filternull_element

filternull_element(values: list) -> list

Source code in src/datajudge/utils.py

def filternull_element(values: list) -> list:
    return [value for value in values if value is not None]

filternull_element_or_tuple_all

filternull_element_or_tuple_all(values: list) -> list

Source code in src/datajudge/utils.py

def filternull_element_or_tuple_all(values: list) -> list:
    return [
        value
        for value in values
        if value is not None
        and not (isinstance(value, tuple) and all(x is None for x in value))
    ]

filternull_element_or_tuple_any

filternull_element_or_tuple_any(values: list) -> list

Source code in src/datajudge/utils.py

def filternull_element_or_tuple_any(values: list) -> list:
    return [
        value
        for value in values
        if value is not None
        and not (isinstance(value, tuple) and any(x is None for x in value))
    ]

filternull_never

filternull_never(values: list) -> list

Source code in src/datajudge/utils.py

def filternull_never(values: list) -> list:
    return values

format_difference

format_difference(n1: float | int, n2: float | int, decimal_separator: bool = True) -> tuple[str, str]

Format and highlight how two numbers differ.

Given two numbers, n1 and n2, return a tuple of two strings, each representing one of the input numbers with the differing part highlighted. Highlighting is done using BBCode-like tags, which are replaced by the formatter.

Examples:

123, 123.0
-> 123, 123[numDiff].0[/numDiff]
122593859432, 122593859432347
-> 122593859432, 122593859432[numDiff]347[/numDiff]

Args: - n1: The first number to compare. - n2: The second number to compare. - decimal_separator: Whether to separate the decimal part of the numbers with commas.

RETURNS	DESCRIPTION
- A tuple of two strings, each representing one of the input numbers with the differing part highlighted.

Source code in src/datajudge/utils.py

def format_difference(
    n1: float | int, n2: float | int, decimal_separator: bool = True
) -> tuple[str, str]:
    """
    Format and highlight how two numbers differ.

    Given two numbers, n1 and n2, return a tuple of two strings,
    each representing one of the input numbers with the differing part highlighted.
    Highlighting is done using BBCode-like tags, which are replaced by the formatter.

    Examples
    --------
        123, 123.0
        -> 123, 123[numDiff].0[/numDiff]
        122593859432, 122593859432347
        -> 122593859432, 122593859432[numDiff]347[/numDiff]

    Args:
    - n1: The first number to compare.
    - n2: The second number to compare.
    - decimal_separator: Whether to separate the decimal part of the numbers with commas.

    Returns
    -------
    - A tuple of two strings, each representing one of the input numbers with the differing part highlighted.
    """
    if decimal_separator:
        s1, s2 = f"{n1:,}", f"{n2:,}"
    else:
        s1, s2 = str(n1), str(n2)

    min_len = min(len(s1), len(s2))
    diff_idx = next(
        (i for i in range(min_len) if s1[i] != s2[i]),
        min_len,
    )

    return (
        f"{s1[:diff_idx]}{_fmt_diff_part(s1, diff_idx)}",
        f"{s2[:diff_idx]}{_fmt_diff_part(s2, diff_idx)}",
    )

output_processor_limit

output_processor_limit(collection: Collection, counts: Collection | None = None, limit: int = 100) -> tuple[Collection, Collection | None]

Limits the collection to the first limit elements.

If the list was shortened, will add a limit+1-th string element, informing the user of the truncation.

The default limit of 100 can be adjusted using functools.partial.

Source code in src/datajudge/utils.py

def output_processor_limit(
    collection: Collection, counts: Collection | None = None, limit: int = 100
) -> tuple[Collection, Collection | None]:
    """
    Limits the collection to the first `limit` elements.

    If the list was shortened, will add a `limit+1`-th string element,
    informing the user of the truncation.

    The default limit of ``100`` can be adjusted using `functools.partial`.
    """
    collection = list(collection)

    ret_collection = collection[:limit]
    ret_counts = None if counts is None else list(counts)[:limit]
    if len(collection) > limit:
        ret_collection.append(
            f"<SHORTENED OUTPUT, displaying the first {limit} / {len(collection)} elements above>"
        )
        if ret_counts is not None:
            ret_counts.append(
                f"<SHORTENED OUTPUT, displaying the first {limit} / {len(collection)} counts above>"
            )

    return ret_collection, ret_counts

output_processor_sort

output_processor_sort(collection: Collection, counts: Collection | None = None) -> tuple[Collection, Collection | None]

Sorts a collection of tuple elements in descending order of their counts.

If ties exist, the ascending order of the elements themselves is used.

If the first element is not instanceof tuple, each element will be transparently packaged into a 1-tuple for processing; this process is not visible to the caller.

Handles None values as described in sort_tuple_none_aware.

Source code in src/datajudge/utils.py

def output_processor_sort(
    collection: Collection, counts: Collection | None = None
) -> tuple[Collection, Collection | None]:
    """
    Sorts a collection of tuple elements in descending order of their counts.

    If ties exist, the ascending order of the elements themselves is used.

    If the first element is not instanceof tuple,
    each element will be transparently packaged into a 1-tuple for processing;
    this process is not visible to the caller.

    Handles ``None`` values as described in ``sort_tuple_none_aware``.
    """
    collection = list(collection)
    if not isinstance(collection[0], tuple):
        # package into a 1 tuple and pass into the method again
        packaged_list = [(elem,) for elem in collection]
        res_main, res_counts = output_processor_sort(packaged_list, counts)
        return [elem[0] for elem in res_main], res_counts

    if counts is None:
        return sort_tuple_none_aware(collection), counts

    if len(collection) != len(counts):
        raise ValueError("collection and counts must have the same length")

    if len(collection) <= 1:
        return collection, counts  # empty or 1 element lists are always sorted

    lst = sort_tuple_none_aware(
        [(-count, *elem) for count, elem in zip(counts, collection)]
    )
    return [elem[1:] for elem in lst], [-elem[0] for elem in lst]

sort_tuple_none_aware

sort_tuple_none_aware(collection: Collection[tuple], ascending=True) -> Collection[tuple]

Stable sort of a collection of tuples.

Each tuple in the collection must have the same length, since they are treated as rows in a table, with elem[0] being the first column, elem[1] the second, etc. for each elem in collection. For sorting, None is considered the same as the default value of the respective column's type.

ints and floats int() and float() yield 0 and 0.0 respectively; for strings, str() yields ''. The constructor is determined by calling type on the first non-None element of the respective column.

Validates that all elements in collection are tuples and that all tuples have the same length.

Source code in src/datajudge/utils.py

def sort_tuple_none_aware(
    collection: Collection[tuple], ascending=True
) -> Collection[tuple]:
    """
    Stable sort of a collection of tuples.

    Each tuple in the collection must have the same length,
    since they are treated as rows in a table,
    with ``elem[0]`` being the first column,
    ``elem[1]`` the second, etc. for each ``elem`` in ``collection``.
    For sorting, ``None`` is considered the same as the default value of the respective column's type.

    ints and floats ``int()`` and ``float()`` yield ``0`` and ``0.0`` respectively; for strings, ``str()`` yields ``''``.
    The constructor is determined by calling ``type`` on the first non-``None`` element of the respective column.

    Validates that all elements in collection are tuples and that all tuples have the same length.
    """
    lst = list(collection)

    if len(lst) <= 1:
        return lst  # empty or 1 element lists are always sorted

    if not all(isinstance(elem, tuple) and len(elem) == len(lst[0]) for elem in lst):
        raise ValueError("all elements must be tuples and have the same length")

    dtypes_each_tupleelement: list[type | None] = [None] * len(lst[0])
    for dtypeidx in range(len(dtypes_each_tupleelement)):
        for elem in lst:
            if elem[dtypeidx] is not None:
                dtypes_each_tupleelement[dtypeidx] = type(elem[dtypeidx])
                break
        else:
            # if all entries are None, just use a constant int() == 0
            dtypes_each_tupleelement[dtypeidx] = int

    def replace_None_with_default(elem):  # noqa: N802
        return tuple(
            ((dtype() if dtype else None) if subelem is None else subelem)
            for dtype, subelem in zip(dtypes_each_tupleelement, elem)
        )

    return sorted(
        lst, key=lambda elem: replace_None_with_default(elem), reverse=not ascending
    )