fastflowtransform.contracts.core

load_contracts

load_contracts(project_dir)

Load all per-table contracts from *.contracts.yml under models/.

Returns:

Type	Description
dict[str, ContractsFileModel]	dict[table_name, ContractsFileModel]
dict[str, ContractsFileModel]	If multiple files define the same table, the last one wins (with a warning).

Source code in src/fastflowtransform/contracts/core.py

def load_contracts(project_dir: Path) -> dict[str, ContractsFileModel]:
    """
    Load all per-table contracts from *.contracts.yml under models/.

    Returns:
        dict[table_name, ContractsFileModel]
        If multiple files define the same `table`, the last one wins (with a warning).
    """
    contracts: dict[str, ContractsFileModel] = {}
    for path in _discover_contract_paths(project_dir):
        cfg = parse_contracts_file(path)

        table = cfg.table
        if table in contracts:
            logger.warning(
                "Multiple contracts for table %r: overriding previous definition with %s",
                table,
                path,
            )
        contracts[table] = cfg

    return contracts

build_contract_tests

build_contract_tests(contracts, *, defaults=None, default_severity='error')

Convert a set of ContractsFileModel objects into a flat list of TestSpec.

defaults is the (optional) project-level defaults section from contracts.yml.

Source code in src/fastflowtransform/contracts/core.py

def build_contract_tests(
    contracts: dict[str, ContractsFileModel],
    *,
    defaults: ContractsDefaultsModel | None = None,
    default_severity: Severity = "error",
) -> list[TestSpec]:
    """
    Convert a set of ContractsFileModel objects into a flat list of TestSpec.

    `defaults` is the (optional) project-level defaults section from contracts.yml.
    """
    if not contracts:
        return []

    all_specs: list[TestSpec] = []
    for table, cfg in contracts.items():
        all_specs.extend(
            _contract_tests_for_table(
                table,
                cfg,
                defaults=defaults,
                default_severity=default_severity,
            )
        )
    return all_specs

load_contract_tests

load_contract_tests(project_dir)

High-level helper used by the CLI:

project_dir -> [TestSpec, ...]

This is what we plug into fft test so contracts become "first-class" tests.

Source code in src/fastflowtransform/contracts/core.py

def load_contract_tests(project_dir: Path) -> list[TestSpec]:
    """
    High-level helper used by the CLI:

        project_dir -> [TestSpec, ...]

    This is what we plug into `fft test` so contracts become "first-class" tests.
    """
    contracts = load_contracts(project_dir)
    if not contracts:
        return []

    project_cfg = _load_project_contracts(project_dir)
    defaults = project_cfg.defaults if project_cfg is not None else None

    return build_contract_tests(contracts, defaults=defaults)