Skip to content

schema Module

Module definine ResultSchema and ArtifactSchema

ResultSchema

Dataclass representing a phase artifact schema

Used to validate a pipeline

Attributes:

Name Type Description
extension str

The artifact file extension (i.e. ".h5" or ".parquet")
structure list[str] | dict[str, Any] | None

The artifact data structure. Can be a list of strings (dataframe columns), a dictionary (HDF5 groups) or None (not specified)
Source code in src/spyral/core/schema.py 
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
76
77
78
79
80
81
82
83
84
85
86
87
class ResultSchema:
    """Dataclass representing a phase artifact schema

    Used to validate a pipeline

    Attributes
    ----------
    extension: str
        The artifact file extension (i.e. ".h5" or ".parquet")
    structure: list[str] | dict[str, Any] | None
        The artifact data structure. Can be a list of strings (dataframe columns),
        a dictionary (HDF5 groups) or None (not specified)

    """

    def __init__(self, json_string: str | None = None, json_file: Path | None = None):
        self.artifacts: None | dict[str, ArtifactSchema] = None
        self.load_schema(json_string, json_file)

    def load_schema(
        self, json_string: str | None = None, json_file: Path | None = None
    ):
        json_data: dict[str, Any]
        if json_string is not None:
            json_data = loads(json_string)
            self.artifacts = {}
        elif json_file is not None:
            with open(json_file) as json_handle:
                json_data = load(json_handle)
            self.artifacts = {}
        else:
            return

        for key, artifact in json_data.items():
            self.artifacts[key] = ArtifactSchema(
                artifact["extension"], artifact["data"]
            )

    def __eq__(self, other: Self) -> bool:
        """Allow comparison of schemas for equivalency

        Parameters
        ----------
        other: ArtifactSchema
            The schema to compare to

        Returns
        -------
        bool
            True if the schemas are equivalent or False if they are different

        """
        return self.artifacts == other.artifacts

__eq__(other)

Allow comparison of schemas for equivalency

Parameters:

Name Type Description Default
other Self

The schema to compare to

 required

Returns:

Type Description
bool

True if the schemas are equivalent or False if they are different
Source code in src/spyral/core/schema.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def __eq__(self, other: Self) -> bool:
    """Allow comparison of schemas for equivalency

    Parameters
    ----------
    other: ArtifactSchema
        The schema to compare to

    Returns
    -------
    bool
        True if the schemas are equivalent or False if they are different

    """
    return self.artifacts == other.artifacts