odcs 0.3.0

Reference implementation of the Open Data Contract Standard (ODCS)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

"""Reference implementation of the Open Data Contract Standard (ODCS)."""

from __future__ import annotations

from importlib.metadata import PackageNotFoundError, version

from pyodcs._native import inspect as _inspect
from pyodcs._native import parse_document as _parse_document
from pyodcs._native import parse_path as _parse_path
from pyodcs._native import upstream_spec_version as _upstream_spec_version
from pyodcs._native import validate_contract as _validate_contract
from pyodcs._native import validate_document as _validate_document

UPSTREAM_SPEC_VERSION = _upstream_spec_version()

try:
    __version__ = version("pyodcs")
except PackageNotFoundError:
    __version__ = "0.2.0"


def parse(content: str | bytes, format: str = "yaml") -> dict:
    """Parse an ODCS document from YAML or JSON text."""
    return _parse_document(content, format)


def parse_file(path: str) -> dict:
    """Parse an ODCS document from a file path."""
    return _parse_path(path)


def validate(contract: dict) -> dict:
    """Validate a parsed data contract."""
    return _validate_contract(contract)


def validate_result(result: dict) -> dict:
    """Merge parse-time and validation diagnostics from a parse result."""
    diagnostics = list(result.get("report", {}).get("diagnostics", []))
    contract = result.get("contract")
    if contract is not None:
        validation = _validate_contract(contract)
        diagnostics.extend(validation.get("diagnostics", []))
    return {"diagnostics": diagnostics}


def parse_and_validate(content: str | bytes, format: str = "yaml") -> dict:
    """Parse and validate an ODCS document in one step."""
    return _validate_document(content, format)


def inspect(contract: dict) -> str:
    """Return a short human-readable contract summary."""
    return _inspect(contract)


def is_valid(report: dict) -> bool:
    """Return True when a diagnostic report contains no error-level diagnostics."""
    return not any(
        diagnostic.get("severity") == "error"
        for diagnostic in report.get("diagnostics", [])
    )


__all__ = [
    "UPSTREAM_SPEC_VERSION",
    "__version__",
    "inspect",
    "is_valid",
    "parse",
    "parse_and_validate",
    "parse_file",
    "validate",
    "validate_result",
]