API Reference

Core

`tollkeeper.Tollkeeper`

Entry point for the write-audit-publish pattern.

Parameters:

Name	Type	Description	Default
`backend`	`Backend`	Storage backend (e.g. `CsvBackend`, `IcebergBackend`).	required

Example::

Tollkeeper(CsvBackend(staging, publish)).table("sales").write(fn).audit([NullCheck("id")]).publish()

Source code in src/tollkeeper/core.py

class Tollkeeper:
    """Entry point for the write-audit-publish pattern.

    Args:
        backend: Storage backend (e.g. ``CsvBackend``, ``IcebergBackend``).

    Example::

        Tollkeeper(CsvBackend(staging, publish)).table("sales").write(fn).audit([NullCheck("id")]).publish()
    """

    def __init__(self, backend: Backend, signal_store: SignalStore | None = None) -> None:
        self._backend = backend
        self._signal_store = signal_store

    def table(self, table: str) -> TollkeeperSession:
        version_ref = self._backend.create_version(table)
        return TollkeeperSession(self._backend, table, version_ref, self._signal_store)

`tollkeeper.TollkeeperSession`

A single write-audit-publish session for one table version.

Created by Tollkeeper.table(). Supports fluent chaining::

Tollkeeper(backend).table("sales").write(fn).audit([checks]).publish()

Source code in src/tollkeeper/core.py

class TollkeeperSession:
    """A single write-audit-publish session for one table version.

    Created by ``Tollkeeper.table()``. Supports fluent chaining::

        Tollkeeper(backend).table("sales").write(fn).audit([checks]).publish()
    """

    def __init__(self, backend: Backend, table: str, version_ref: str, signal_store: SignalStore | None = None) -> None:
        self._backend = backend
        self._table = table
        self._version_ref = version_ref
        self._report = CheckReport()
        self._published = False
        self._rolled_back = False
        self._audited = False
        self._signal_store = signal_store

    def __del__(self) -> None:
        if not self._published and not self._rolled_back:
            import logging

            logging.getLogger(__name__).warning(
                "TollkeeperSession for %s was never published or rolled back — rolling back", self._table
            )
            try:
                self._backend.rollback_version(self._table, self._version_ref)
            except Exception:
                pass
            self._rolled_back = True

    def __enter__(self) -> TollkeeperSession:
        return self

    def __exit__(self, exc_type: type | None, exc_val: BaseException | None, exc_tb: object) -> None:
        if not self._published and not self._rolled_back:
            try:
                self._backend.rollback_version(self._table, self._version_ref)
            except Exception:
                pass
            self._rolled_back = True

    @property
    def ref(self) -> str:
        return self._version_ref

    @property
    def report(self) -> CheckReport:
        return self._report

    def write(self, fn: Callable[[str], None]) -> TollkeeperSession:
        """Execute the write function against the staged version.

        Args:
            fn: Callable that receives the version reference and writes data to it.

        Raises:
            Exception: Re-raises any exception from ``fn`` after rolling back the staged version.
        """
        try:
            fn(self._version_ref)
        except Exception:
            self._backend.rollback_version(self._table, self._version_ref)
            self._rolled_back = True
            raise
        return self

    def audit(
        self,
        checks: list[BaseCheck],
        *,
        on_failure: str = "stop",
        on_notify: Callable[[str, str, list[CheckResult]], None] | None = None,
        execution_ctx: dict | None = None,
        conn: Any | None = None,
    ) -> TollkeeperSession:
        if on_failure not in ("stop", "continue"):
            raise ValueError("on_failure must be 'stop' or 'continue'")

        if self._signal_store:
            self._signal_store.delete(self._table, execution_ctx)

        self._report.results.extend(check.run(self._version_ref, conn=conn) for check in checks)
        self._audited = True

        if self._report.failed:
            if on_notify:
                on_notify(self._table, self._version_ref, self._report.failed)
            if on_failure == "stop":
                try:
                    self._backend.rollback_version(self._table, self._version_ref)
                finally:
                    self._rolled_back = True
                    raise AuditFailedError(self._table, self._version_ref, self._report.failed)

        if self._signal_store and self._report.passed:
            from datetime import datetime

            from tollkeeper.signals.base import Signal

            self._signal_store.write(
                Signal(
                    table_name=self._table,
                    execution_ctx=execution_ctx or {},
                    status="passed",
                    execution_ts=datetime.now(),
                    check_summary=f"{len(self._report.results)} checks passed",
                )
            )

        return self

    def publish(self) -> TollkeeperSession:
        """Promote the staged version to production.

        Raises:
            RuntimeError: If the session was already rolled back.
            RuntimeError: If ``audit()`` has not been called.
        """
        if self._rolled_back:
            raise RuntimeError("Cannot publish a rolled-back session")
        if not self._audited:
            raise RuntimeError("Cannot publish without running audit first")
        if not self._published:
            self._backend.publish_version(self._table, self._version_ref)
            self._published = True
        return self

    def rollback(self) -> TollkeeperSession:
        """Discard the staged version without publishing.

        Raises:
            RuntimeError: If the session was already published.
        """
        if self._published:
            raise RuntimeError("Cannot rollback a published session")
        if not self._rolled_back:
            self._backend.rollback_version(self._table, self._version_ref)
            self._rolled_back = True
        return self

`publish()`

Promote the staged version to production.

Raises:

Type	Description
`RuntimeError`	If the session was already rolled back.
`RuntimeError`	If `audit()` has not been called.

Source code in src/tollkeeper/core.py

def publish(self) -> TollkeeperSession:
    """Promote the staged version to production.

    Raises:
        RuntimeError: If the session was already rolled back.
        RuntimeError: If ``audit()`` has not been called.
    """
    if self._rolled_back:
        raise RuntimeError("Cannot publish a rolled-back session")
    if not self._audited:
        raise RuntimeError("Cannot publish without running audit first")
    if not self._published:
        self._backend.publish_version(self._table, self._version_ref)
        self._published = True
    return self

`rollback()`

Discard the staged version without publishing.

Raises:

Type	Description
`RuntimeError`	If the session was already published.

Source code in src/tollkeeper/core.py

def rollback(self) -> TollkeeperSession:
    """Discard the staged version without publishing.

    Raises:
        RuntimeError: If the session was already published.
    """
    if self._published:
        raise RuntimeError("Cannot rollback a published session")
    if not self._rolled_back:
        self._backend.rollback_version(self._table, self._version_ref)
        self._rolled_back = True
    return self

`write(fn)`

Execute the write function against the staged version.

Parameters:

Name	Type	Description	Default
`fn`	`Callable[[str], None]`	Callable that receives the version reference and writes data to it.	required

Raises:

Type	Description
`Exception`	Re-raises any exception from `fn` after rolling back the staged version.

Source code in src/tollkeeper/core.py

def write(self, fn: Callable[[str], None]) -> TollkeeperSession:
    """Execute the write function against the staged version.

    Args:
        fn: Callable that receives the version reference and writes data to it.

    Raises:
        Exception: Re-raises any exception from ``fn`` after rolling back the staged version.
    """
    try:
        fn(self._version_ref)
    except Exception:
        self._backend.rollback_version(self._table, self._version_ref)
        self._rolled_back = True
        raise
    return self

`tollkeeper.CheckReport` `dataclass`

Aggregated results from an audit pass.

Attributes:

Name	Type	Description
`results`	`list[CheckResult]`	List of individual `CheckResult` objects.

Source code in src/tollkeeper/core.py

@dataclass
class CheckReport:
    """Aggregated results from an audit pass.

    Attributes:
        results: List of individual ``CheckResult`` objects.
    """

    results: list[CheckResult] = field(default_factory=list)

    @property
    def passed(self) -> bool:
        return all(r.passed for r in self.results)

    @property
    def failed(self) -> list[CheckResult]:
        return [r for r in self.results if not r.passed]

`tollkeeper.AuditFailedError`

Bases: Exception

Raised when a hard DQ audit (on_failure="stop") finds violations.

Source code in src/tollkeeper/core.py

class AuditFailedError(Exception):
    """Raised when a hard DQ audit (``on_failure="stop"``) finds violations."""

    def __init__(self, table: str, version_ref: str, failed: list[CheckResult]) -> None:
        self.table = table
        self.version_ref = version_ref
        self.failed = failed
        names = ", ".join(r.check_name for r in failed)
        super().__init__(f"Audit failed for {table}@{version_ref}: {names}")

Backends

`tollkeeper.backends.base.Backend`

Bases: ABC

Abstract base for Tollkeeper storage backends.

A backend manages the lifecycle of a versioned table: creating an isolated staging area, promoting it to production, or rolling it back. Implement this to add support for a new storage layer (CSV files, Iceberg, Delta, etc.).

Source code in src/tollkeeper/backends/base.py

class Backend(ABC):
    """Abstract base for Tollkeeper storage backends.

    A backend manages the lifecycle of a versioned table: creating an isolated
    staging area, promoting it to production, or rolling it back. Implement this
    to add support for a new storage layer (CSV files, Iceberg, Delta, etc.).
    """

    @abstractmethod
    def create_version(self, table: str) -> str:
        """Create an isolated staging version and return a version reference.

        The reference is backend-specific: a file path, branch name, snapshot ID, etc.
        The caller writes data to whatever the reference points at.

        Args:
            table: Logical table name (e.g. ``"sales"``).

        Returns:
            An opaque version reference string passed to subsequent methods.
        """

    @abstractmethod
    def publish_version(self, table: str, version_ref: str) -> None:
        """Promote the staged version to production.

        Args:
            table: Logical table name.
            version_ref: Reference returned by ``create_version``.
        """

    @abstractmethod
    def rollback_version(self, table: str, version_ref: str) -> None:
        """Discard a staged version without publishing.

        Args:
            table: Logical table name.
            version_ref: Reference returned by ``create_version``.
        """

`create_version(table)` `abstractmethod`

Create an isolated staging version and return a version reference.

The reference is backend-specific: a file path, branch name, snapshot ID, etc. The caller writes data to whatever the reference points at.

Parameters:

Name	Type	Description	Default
`table`	`str`	Logical table name (e.g. `"sales"`).	required

Returns:

Type	Description
`str`	An opaque version reference string passed to subsequent methods.

Source code in src/tollkeeper/backends/base.py

@abstractmethod
def create_version(self, table: str) -> str:
    """Create an isolated staging version and return a version reference.

    The reference is backend-specific: a file path, branch name, snapshot ID, etc.
    The caller writes data to whatever the reference points at.

    Args:
        table: Logical table name (e.g. ``"sales"``).

    Returns:
        An opaque version reference string passed to subsequent methods.
    """

`publish_version(table, version_ref)` `abstractmethod`

Promote the staged version to production.

Parameters:

Name	Type	Description	Default
`table`	`str`	Logical table name.	required
`version_ref`	`str`	Reference returned by `create_version`.	required

Source code in src/tollkeeper/backends/base.py

@abstractmethod
def publish_version(self, table: str, version_ref: str) -> None:
    """Promote the staged version to production.

    Args:
        table: Logical table name.
        version_ref: Reference returned by ``create_version``.
    """

`rollback_version(table, version_ref)` `abstractmethod`

Discard a staged version without publishing.

Parameters:

Name	Type	Description	Default
`table`	`str`	Logical table name.	required
`version_ref`	`str`	Reference returned by `create_version`.	required

Source code in src/tollkeeper/backends/base.py

@abstractmethod
def rollback_version(self, table: str, version_ref: str) -> None:
    """Discard a staged version without publishing.

    Args:
        table: Logical table name.
        version_ref: Reference returned by ``create_version``.
    """

`tollkeeper.backends.csv.CsvBackend`

Bases: Backend

Tollkeeper backend for local CSV files.

Stages data as a temporary CSV in staging_dir, then copies to publish_dir on publish or deletes on rollback.

Parameters:

Name	Type	Description	Default
`staging_dir`	`Path`	Directory for temporary staging files.	required
`publish_dir`	`Path`	Directory where final published CSVs land.	required

Example::

backend = CsvBackend(staging_dir=Path("/tmp/tollkeeper"), publish_dir=Path("/data/output"))
Tollkeeper(backend).table("sales").write(lambda ref: shutil.copy(src, ref)).audit([...]).publish()

Source code in src/tollkeeper/backends/csv.py

class CsvBackend(Backend):
    """Tollkeeper backend for local CSV files.

    Stages data as a temporary CSV in ``staging_dir``, then copies to
    ``publish_dir`` on publish or deletes on rollback.

    Args:
        staging_dir: Directory for temporary staging files.
        publish_dir: Directory where final published CSVs land.

    Example::

        backend = CsvBackend(staging_dir=Path("/tmp/tollkeeper"), publish_dir=Path("/data/output"))
        Tollkeeper(backend).table("sales").write(lambda ref: shutil.copy(src, ref)).audit([...]).publish()
    """

    def __init__(self, staging_dir: Path, publish_dir: Path) -> None:
        self._staging_dir = staging_dir
        self._publish_dir = publish_dir

    def create_version(self, table: str) -> str:
        self._staging_dir.mkdir(parents=True, exist_ok=True)
        staging_path = self._staging_dir / f"{table}.tollkeeper-{uuid4().hex[:8]}.csv"
        return str(staging_path)

    def publish_version(self, table: str, version_ref: str) -> None:
        self._publish_dir.mkdir(parents=True, exist_ok=True)
        _move(Path(version_ref), self._publish_dir / f"{table}.csv")

    def rollback_version(self, table: str, version_ref: str) -> None:
        Path(version_ref).unlink(missing_ok=True)

    def cleanup_staging(self, max_age_seconds: float = 3600) -> list[Path]:
        """Remove Tollkeeper staging files older than ``max_age_seconds``.

        Args:
            max_age_seconds: Maximum age in seconds before a staging file is considered orphaned.

        Returns:
            List of removed file paths.
        """
        if not self._staging_dir.exists():
            return []
        cutoff = time.time() - max_age_seconds
        removed: list[Path] = []
        for p in self._staging_dir.glob("*.tollkeeper-*.csv"):
            if p.stat().st_mtime < cutoff:
                p.unlink()
                removed.append(p)
        return removed

`cleanup_staging(max_age_seconds=3600)`

Remove Tollkeeper staging files older than max_age_seconds.

Parameters:

Name	Type	Description	Default
`max_age_seconds`	`float`	Maximum age in seconds before a staging file is considered orphaned.	`3600`

Returns:

Type	Description
`list[Path]`	List of removed file paths.

Source code in src/tollkeeper/backends/csv.py

def cleanup_staging(self, max_age_seconds: float = 3600) -> list[Path]:
    """Remove Tollkeeper staging files older than ``max_age_seconds``.

    Args:
        max_age_seconds: Maximum age in seconds before a staging file is considered orphaned.

    Returns:
        List of removed file paths.
    """
    if not self._staging_dir.exists():
        return []
    cutoff = time.time() - max_age_seconds
    removed: list[Path] = []
    for p in self._staging_dir.glob("*.tollkeeper-*.csv"):
        if p.stat().st_mtime < cutoff:
            p.unlink()
            removed.append(p)
    return removed

`tollkeeper.backends.iceberg.IcebergBackend`

Bases: Backend

Tollkeeper backend for Apache Iceberg tables using branch-based isolation.

Creates a temporary branch for staging writes, then fast-forwards main on publish or drops the branch on rollback.

Parameters:

Name	Type	Description	Default
`catalog`	`Catalog`	A PyIceberg `Catalog` instance.	required

Example::

from pyiceberg.catalog.sql import SqlCatalog
catalog = SqlCatalog("default", warehouse="/tmp/warehouse", uri="sqlite:///catalog.db")
backend = IcebergBackend(catalog)
Tollkeeper(backend).table("db.sales").write(lambda ref: table.append(df, branch=ref)).audit([...]).publish()

Source code in src/tollkeeper/backends/iceberg.py

class IcebergBackend(Backend):
    """Tollkeeper backend for Apache Iceberg tables using branch-based isolation.

    Creates a temporary branch for staging writes, then fast-forwards ``main``
    on publish or drops the branch on rollback.

    Args:
        catalog: A PyIceberg ``Catalog`` instance.

    Example::

        from pyiceberg.catalog.sql import SqlCatalog
        catalog = SqlCatalog("default", warehouse="/tmp/warehouse", uri="sqlite:///catalog.db")
        backend = IcebergBackend(catalog)
        Tollkeeper(backend).table("db.sales").write(lambda ref: table.append(df, branch=ref)).audit([...]).publish()
    """

    def __init__(self, catalog: Catalog) -> None:
        self._catalog = catalog

    def create_version(self, table: str) -> str:
        """Create a Tollkeeper branch on the Iceberg table.

        Args:
            table: Fully qualified Iceberg table identifier (e.g. ``"db.sales"``).

        Returns:
            Branch name (e.g. ``"tollkeeper-a1b2c3d4"``). Pass this as the ``branch``
            parameter when writing via PyIceberg.
        """
        iceberg_table = self._catalog.load_table(table)
        branch_name = f"tollkeeper-{uuid4().hex[:8]}"
        snapshot = iceberg_table.current_snapshot()
        if snapshot is None:
            raise RuntimeError(f"Table {table} has no snapshots — write initial data before using Tollkeeper")
        iceberg_table.manage_snapshots().create_branch(snapshot.snapshot_id, branch_name).commit()
        return branch_name

    def publish_version(self, table: str, version_ref: str) -> None:
        """Fast-forward ``main`` to the branch snapshot, then remove the branch.

        Args:
            table: Fully qualified Iceberg table identifier.
            version_ref: Branch name returned by ``create_version``.
        """
        iceberg_table = self._catalog.load_table(table)
        iceberg_table.manage_snapshots().set_current_snapshot(ref_name=version_ref).commit()
        iceberg_table.manage_snapshots().remove_branch(version_ref).commit()

    def rollback_version(self, table: str, version_ref: str) -> None:
        """Drop the Tollkeeper branch without publishing.

        Args:
            table: Fully qualified Iceberg table identifier.
            version_ref: Branch name returned by ``create_version``.
        """
        iceberg_table = self._catalog.load_table(table)
        try:
            iceberg_table.manage_snapshots().remove_branch(version_ref).commit()
        except Exception:
            pass

`create_version(table)`

Create a Tollkeeper branch on the Iceberg table.

Parameters:

Name	Type	Description	Default
`table`	`str`	Fully qualified Iceberg table identifier (e.g. `"db.sales"`).	required

Returns:

Type	Description
`str`	Branch name (e.g. `"tollkeeper-a1b2c3d4"`). Pass this as the `branch`
`str`	parameter when writing via PyIceberg.

Source code in src/tollkeeper/backends/iceberg.py

def create_version(self, table: str) -> str:
    """Create a Tollkeeper branch on the Iceberg table.

    Args:
        table: Fully qualified Iceberg table identifier (e.g. ``"db.sales"``).

    Returns:
        Branch name (e.g. ``"tollkeeper-a1b2c3d4"``). Pass this as the ``branch``
        parameter when writing via PyIceberg.
    """
    iceberg_table = self._catalog.load_table(table)
    branch_name = f"tollkeeper-{uuid4().hex[:8]}"
    snapshot = iceberg_table.current_snapshot()
    if snapshot is None:
        raise RuntimeError(f"Table {table} has no snapshots — write initial data before using Tollkeeper")
    iceberg_table.manage_snapshots().create_branch(snapshot.snapshot_id, branch_name).commit()
    return branch_name

`publish_version(table, version_ref)`

Fast-forward main to the branch snapshot, then remove the branch.

Parameters:

Name	Type	Description	Default
`table`	`str`	Fully qualified Iceberg table identifier.	required
`version_ref`	`str`	Branch name returned by `create_version`.	required

Source code in src/tollkeeper/backends/iceberg.py

def publish_version(self, table: str, version_ref: str) -> None:
    """Fast-forward ``main`` to the branch snapshot, then remove the branch.

    Args:
        table: Fully qualified Iceberg table identifier.
        version_ref: Branch name returned by ``create_version``.
    """
    iceberg_table = self._catalog.load_table(table)
    iceberg_table.manage_snapshots().set_current_snapshot(ref_name=version_ref).commit()
    iceberg_table.manage_snapshots().remove_branch(version_ref).commit()

`rollback_version(table, version_ref)`

Drop the Tollkeeper branch without publishing.

Parameters:

Name	Type	Description	Default
`table`	`str`	Fully qualified Iceberg table identifier.	required
`version_ref`	`str`	Branch name returned by `create_version`.	required

Source code in src/tollkeeper/backends/iceberg.py

def rollback_version(self, table: str, version_ref: str) -> None:
    """Drop the Tollkeeper branch without publishing.

    Args:
        table: Fully qualified Iceberg table identifier.
        version_ref: Branch name returned by ``create_version``.
    """
    iceberg_table = self._catalog.load_table(table)
    try:
        iceberg_table.manage_snapshots().remove_branch(version_ref).commit()
    except Exception:
        pass

Checks

`tollkeeper.checks.base.BaseCheck`

Bases: ABC

Abstract base for data quality checks.

Subclass this to create checks using any engine (Polars, Pandas, Presto, etc.). The run method receives a version reference (e.g. a file path or table name) and returns a CheckResult.

Source code in src/tollkeeper/checks/base.py

class BaseCheck(ABC):
    """Abstract base for data quality checks.

    Subclass this to create checks using any engine (Polars, Pandas, Presto, etc.).
    The ``run`` method receives a version reference (e.g. a file path or table name)
    and returns a ``CheckResult``.
    """

    @property
    def name(self) -> str:
        return self.__class__.__name__

    @abstractmethod
    def run(self, version_ref: str, *, conn: Any | None = None) -> CheckResult:
        """Execute the check against the given version reference.

        Args:
            version_ref: Backend-specific reference to the staged data.
            conn: Optional engine connection for remote check execution.
                When ``None``, checks run in-process.

        Returns:
            A ``CheckResult`` indicating pass/fail with details.
        """

`run(version_ref, *, conn=None)` `abstractmethod`

Execute the check against the given version reference.

Parameters:

Name	Type	Description	Default
`version_ref`	`str`	Backend-specific reference to the staged data.	required
`conn`	`Any \| None`	Optional engine connection for remote check execution. When `None`, checks run in-process.	`None`

Returns:

Type	Description
`CheckResult`	A `CheckResult` indicating pass/fail with details.

Source code in src/tollkeeper/checks/base.py

@abstractmethod
def run(self, version_ref: str, *, conn: Any | None = None) -> CheckResult:
    """Execute the check against the given version reference.

    Args:
        version_ref: Backend-specific reference to the staged data.
        conn: Optional engine connection for remote check execution.
            When ``None``, checks run in-process.

    Returns:
        A ``CheckResult`` indicating pass/fail with details.
    """

`tollkeeper.checks.base.CheckResult` `dataclass`

Result of a single DQ check execution.

Attributes:

Name	Type	Description
`check_name`	`str`	Name of the check that produced this result.
`passed`	`bool`	Whether the check passed.
`details`	`str`	Human-readable details (e.g. `"3 nulls in 'id'"`).

Source code in src/tollkeeper/checks/base.py

@dataclass
class CheckResult:
    """Result of a single DQ check execution.

    Attributes:
        check_name: Name of the check that produced this result.
        passed: Whether the check passed.
        details: Human-readable details (e.g. ``"3 nulls in 'id'"``).
    """

    check_name: str
    passed: bool
    details: str = ""

`tollkeeper.checks.polars.NullCheck`

Bases: BaseCheck

Fails if the specified column contains any null values.

Parameters:

Name	Type	Description	Default
`column`	`str`	Column name to check for nulls.	required

Source code in src/tollkeeper/checks/polars.py

class NullCheck(BaseCheck):
    """Fails if the specified column contains any null values.

    Args:
        column: Column name to check for nulls.
    """

    def __init__(self, column: str) -> None:
        self._column = column

    def run(self, version_ref: str, *, conn: Any | None = None) -> CheckResult:
        df = pl.scan_csv(version_ref).collect()
        null_count = df[self._column].null_count()
        return CheckResult(
            check_name=self.name,
            passed=null_count == 0,
            details=f"{null_count} nulls in '{self._column}'",
        )

`tollkeeper.checks.polars.RowCountCheck`

Bases: BaseCheck

Fails if the row count is below the minimum threshold.

Parameters:

Name	Type	Description	Default
`min_rows`	`int`	Minimum number of rows required to pass.	required

Source code in src/tollkeeper/checks/polars.py

class RowCountCheck(BaseCheck):
    """Fails if the row count is below the minimum threshold.

    Args:
        min_rows: Minimum number of rows required to pass.
    """

    def __init__(self, min_rows: int) -> None:
        self._min_rows = min_rows

    def run(self, version_ref: str, *, conn: Any | None = None) -> CheckResult:
        df = pl.scan_csv(version_ref).collect()
        count = len(df)
        return CheckResult(
            check_name=self.name,
            passed=count >= self._min_rows,
            details=f"{count} rows, minimum {self._min_rows}",
        )

`tollkeeper.checks.polars.ExpressionCheck`

Bases: BaseCheck

Fails if any row does not satisfy a Polars expression.

Parameters:

Name	Type	Description	Default
`name`	`str`	Check name (used in reports).	required
`expr`	`Expr`	A Polars expression that evaluates to boolean per row.	required

Example::

ExpressionCheck("positive_age", pl.col("age") > 0)

Source code in src/tollkeeper/checks/polars.py

class ExpressionCheck(BaseCheck):
    """Fails if any row does not satisfy a Polars expression.

    Args:
        name: Check name (used in reports).
        expr: A Polars expression that evaluates to boolean per row.

    Example::

        ExpressionCheck("positive_age", pl.col("age") > 0)
    """

    def __init__(self, name: str, expr: pl.Expr) -> None:
        self._name = name
        self._expr = expr

    @property
    def name(self) -> str:
        return self._name

    def run(self, version_ref: str, *, conn: Any | None = None) -> CheckResult:
        df = pl.scan_csv(version_ref).collect()
        violations = df.filter(~self._expr)
        return CheckResult(
            check_name=self.name,
            passed=len(violations) == 0,
            details=f"{len(violations)} rows violate '{self._name}'",
        )

`tollkeeper.checks.polars.SqlCheck`

Bases: BaseCheck

Fails if any row does not satisfy a SQL WHERE condition.

Uses Polars SQLContext to evaluate the condition. The staged data is registered as a table named data.

Parameters:

Name	Type	Description	Default
`name`	`str`	Check name (used in reports).	required
`condition`	`str`	SQL WHERE clause (e.g. `"age > 0 AND name IS NOT NULL"`).	required

Example::

SqlCheck("valid_age", "age > 0 AND age < 150")

Source code in src/tollkeeper/checks/polars.py

class SqlCheck(BaseCheck):
    """Fails if any row does not satisfy a SQL WHERE condition.

    Uses Polars ``SQLContext`` to evaluate the condition. The staged data
    is registered as a table named ``data``.

    Args:
        name: Check name (used in reports).
        condition: SQL WHERE clause (e.g. ``"age > 0 AND name IS NOT NULL"``).

    Example::

        SqlCheck("valid_age", "age > 0 AND age < 150")
    """

    def __init__(self, name: str, condition: str) -> None:
        self._name = name
        self._condition = condition

    @property
    def name(self) -> str:
        return self._name

    def run(self, version_ref: str, *, conn: Any | None = None) -> CheckResult:
        df = pl.scan_csv(version_ref).collect()
        ctx = pl.SQLContext({"data": df})
        violations = ctx.execute(f"SELECT * FROM data WHERE NOT ({self._condition})").collect()  # nosec B608 - Polars in-memory SQL, not a real DB
        return CheckResult(
            check_name=self.name,
            passed=len(violations) == 0,
            details=f"{len(violations)} rows violate '{self._name}'",
        )

`tollkeeper.checks.polars.UniqueCheck`

Bases: BaseCheck

Fails if any combination of the given columns has duplicate rows.

Parameters:

Name	Type	Description	Default
`columns`	`list[str]`	List of column names that should form a unique key.	required

Example::

UniqueCheck(["region", "date"])

Source code in src/tollkeeper/checks/polars.py

class UniqueCheck(BaseCheck):
    """Fails if any combination of the given columns has duplicate rows.

    Args:
        columns: List of column names that should form a unique key.

    Example::

        UniqueCheck(["region", "date"])
    """

    def __init__(self, columns: list[str]) -> None:
        self._columns = columns

    def run(self, version_ref: str, *, conn: Any | None = None) -> CheckResult:
        df = pl.scan_csv(version_ref).collect()
        duplicates = df.group_by(self._columns).len().filter(pl.col("len") > 1)
        return CheckResult(
            check_name=self.name,
            passed=len(duplicates) == 0,
            details=f"{len(duplicates)} duplicate groups on {self._columns}",
        )