fastflowtransform.cli.bootstrap

configure_executor_contracts

configure_executor_contracts(project_dir, executor)

Load contracts from project_dir and attach them to the executor (if supported).

Mirrors the behaviour in fft run: parse per-table contracts and the project-level contracts.yml; on parse errors, log a warning and continue without contracts.

Source code in src/fastflowtransform/cli/bootstrap.py

def configure_executor_contracts(project_dir: Path, executor: BaseExecutor | None) -> None:
    """
    Load contracts from project_dir and attach them to the executor (if supported).

    Mirrors the behaviour in `fft run`: parse per-table contracts and the
    project-level contracts.yml; on parse errors, log a warning and continue
    without contracts.
    """
    if executor is None or not hasattr(executor, "configure_contracts"):
        return

    try:
        contracts_by_table = load_contracts(project_dir)
        project_contracts = _load_project_contracts(project_dir)
    except Exception as exc:
        warn(f"[contracts] Failed to load contracts from {project_dir}: {exc}")
        contracts_by_table = {}
        project_contracts = None

    with suppress(Exception):
        executor.configure_contracts(contracts_by_table, project_contracts)