Api

def starter_tests_from_samples(
    samples: Iterable[FrameT],
    margin: float = 1,
) -> list[dict]:
    """From a list of supported dataframes, produce a list of passing tests.

    Margin is worth explaining here, as it's the amount of *extra margin* tests
    are given, based on standard deviation. If three dataframes are given, with
    "column_a" values that have maximums of 1, 2 and 3, rather than creating a
    test that the maximum should be 3, Wimsey will give a degree of margin.

    By default this will be 1 standard deviation (of the maximum values). So for
    the above example, Wimsey will test for a maximum for 4. This can be tuned with
    the 'margin' keyword.
    """
    df_samples: list[dict[str, Any]] = _profile_from_samples(samples)
    return _starter_tests_from_sample_describes(df_samples, margin)

def starter_tests_from_sampling(
    df: FrameT,
    samples: int = 100,
    n: int | None = None,
    fraction: int | None = None,
    margin: float = 1,
) -> list[dict]:
    """Build out starter tests from a dataframe.

    From a supported dataframe, produce a list of passing tests by sampling the
    dataframe. Note n *or* fraction should be given, but not both. Keyword `samples`
    relates to the *number of samples* to take, whereas `n` relates to the *size of
    a given sample*.

    Margin is worth explaining here, as it's the amount of *extra margin* tests
    are given, based on standard deviation. If three dataframes are given, with
    "column_a" values that have maximums of 1, 2 and 3, rather than creating a
    test that the maximum should be 3, Wimsey will give a degree of margin.

    By default this will be 1 standard deviation (of the maximum values). So for
    the above example, Wimsey will test for a maximum for 4. This can be tuned with
    the 'margin' keyword.
    """
    df_samples = _profile_from_sampling(df, samples, n, fraction)
    return _starter_tests_from_sample_describes(df_samples, margin)

def test(
    df: FrameT,
    contract: str | list[dict] | dict | list[GeneratedTest],
    storage_options: dict | None = None,
) -> FinalResult:
    """Test a dataframe against a data contract.

    Carry out tests on dataframe and return results. This will *not* raise
    an exception on test failure, and will instead return a 'final_result'
    object, with a boolean 'success' field, and a detailed list of individual
    tests.

    If you want to halt processing in the event of a data contract failure,
    see `validate` function.
    """
    tests = (
        _read_config(path=contract, storage_options=storage_options)
        if isinstance(contract, str)
        else _collect_tests(contract)
    )
    return _run_all_tests(df, tests)

def test_or_build(
    df: FrameT,
    contract: str,
    samples: int = 100,
    n: int | None = None,
    fraction: int | None = None,
    margin: float = 1,
    storage_options: dict | None = None,
) -> FinalResult:
    """Validate dataframe if test file exists, otherwise make one.

    Test file will be generated from multple samples of dataset.

    Will fall back to starter_tests_from_sampling (a list samples is not possible with
    only one dataframe), see *starter_tests_from_sampling* and *save_starter_tests_from_sampling*
    for more details on use of keyword arguments aside from df, contract and storage_options.
    """
    try:
        return test(df, contract=contract, storage_options=storage_options)
    except FileNotFoundError:
        save_starter_tests_from_sampling(
            path=contract,
            df=df,
            samples=samples,
            n=n,
            fraction=fraction,
            margin=margin,
        )
        return FinalResult(True, [Result("tests-generated", True)])

def validate(
    df: FrameT,
    contract: str | list[dict] | dict | list[GeneratedTest],
    storage_options: dict | None = None,
) -> FrameT:
    """Validate a dataframe against a data contract.

    Carry out tests on dataframe, returning original dataframe if tests are
    successful, and raising a DataValidationException in case of failure.
    """
    results = test(
        df=df,
        contract=contract,
        storage_options=storage_options,
    )
    if not results.success:
        failures: list[str] = [
            f"{i.name} (unexpected: {i.unexpected})" for i in results.results if not i.success
        ]
        newline = "\n - "
        msg = f"At least one test failed:\n - {newline.join(failures)}"
        raise DataValidationError(msg)
    return df

def validate_or_build(
    df: FrameT,
    contract: str,
    samples: int = 100,
    n: int | None = None,
    fraction: int | None = None,
    margin: float = 1,
    storage_options: dict | None = None,
) -> FrameT:
    """Validate dataframe if test file exists, otherwise make one.

    Test file will be generated from multple samples of dataset.

    Will fall back to starter_tests_from_sampling (a list samples is not possible with
    only one dataframe), see *starter_tests_from_sampling* and *save_starter_tests_from_sampling*
    for more details on use of keyword arguments aside from df, contract and storage_options.
    """
    try:
        return validate(df, contract=contract, storage_options=storage_options)
    except FileNotFoundError:
        save_starter_tests_from_sampling(
            path=contract,
            df=df,
            samples=samples,
            n=n,
            fraction=fraction,
            margin=margin,
        )
        return df