Skip to content

Main Module

datajoint_file_validator.main.validate(target, manifest, verbose=False, raise_err=False, format='table')

Validate a target against a manifest.

Parameters

target : PathLike | Snapshot A path to a file or directory, or an instance of a Snapshot object. manifest : PathLike | Manifest Path to a manifest file, or an instance of a Manifest object. verbose : bool Print verbose output. raise_err : bool Raise an error if validation fails. format : str Format for error report. One of "table", "yaml", or "json".

Returns

result : dict A dictionary with the validation result.

Source code in datajoint_file_validator/main.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def validate(
    target: Union[Snapshot, PathLike],
    manifest: Union[Manifest, PathLike],
    verbose=False,
    raise_err=False,
    format="table",
) -> Tuple[bool, ErrorReport]:
    """
    Validate a target against a manifest.

    Parameters
    ----------
    target : PathLike | Snapshot
        A path to a file or directory, or an instance of a Snapshot object.
    manifest : PathLike | Manifest
        Path to a manifest file, or an instance of a Manifest object.
    verbose : bool
        Print verbose output.
    raise_err : bool
        Raise an error if validation fails.
    format : str
        Format for error report. One of "table", "yaml", or "json".

    Returns
    -------
    result : dict
        A dictionary with the validation result.
    """
    # Infer how to fetch manifest
    if isinstance(manifest, Manifest):
        mani = manifest
    else:
        mani = Manifest.from_yaml(find_manifest(manifest))

    # Infer how to create snapshot
    if isinstance(target, str):
        target = create_snapshot(target)

    return validate_snapshot(
        target, mani, verbose=verbose, raise_err=raise_err, format=format
    )

datajoint_file_validator.main.validate_snapshot(snapshot, manifest, verbose=False, raise_err=False, format='table')

Validate a snapshot against a manifest.

Parameters

snapshot : Snapshot A snapshot dictionary. manifest_path : PathLike Path to a manifest file. verbose : bool Print verbose output. raise_err : bool Raise an error if validation fails. format : str Format for error report. One of "table", "yaml", or "json".

Returns

result : dict A dictionary with the validation result.

Source code in datajoint_file_validator/main.py
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def validate_snapshot(
    snapshot: Snapshot,
    manifest: Manifest,
    verbose=False,
    raise_err=False,
    format="table",
) -> Tuple[bool, ErrorReport]:
    """
    Validate a snapshot against a manifest.

    Parameters
    ----------
    snapshot : Snapshot
        A snapshot dictionary.
    manifest_path : PathLike
        Path to a manifest file.
    verbose : bool
        Print verbose output.
    raise_err : bool
        Raise an error if validation fails.
    format : str
        Format for error report. One of "table", "yaml", or "json".

    Returns
    -------
    result : dict
        A dictionary with the validation result.
    """
    results: List[Dict[str, ValidationResult]] = list(
        map(lambda rule: rule.validate(snapshot), manifest.rules)
    )
    success = all(map(lambda result: all(result.values()), results))

    # Generate error report
    error_report = []
    for rule, result in zip(manifest.rules, results):
        for constraint, valresult in result.items():
            if valresult.status:
                continue
            error_report.append(
                {
                    "rule": rule.id,
                    "rule_description": rule.description,
                    "constraint_id": constraint,
                    "constraint_value": valresult.context["constraint"].val,
                    "errors": valresult.message,
                }
            )
    if verbose and not success:
        rprint("Validation failed with the following errors:", file=sys.stderr)
        if format == "table":
            table = table_from_report(error_report)
            console = Console()
            console.print(table)
        elif format == "yaml":
            rprint(yaml.dump(error_report))
        elif format == "json":
            rprint(json.dumps(error_report, indent=2))
        else:
            raise ValueError(f"Unsupported format: {format}")
    if raise_err and not success:
        raise DJFileValidatorError("Validation failed.")

    return success, error_report