Skip to content

Checks

SodaCL check block

SodaCLCheck

Bases: Block

This block represents a SodaCL check that can be used when running Soda scans.

Parameters:

Name Type Description Default
sodacl_yaml_path str

Absolute path of the Soda Checks file.

 required
sodacl_yaml_str str

Optional YAML string containing the Soda Checks details. If provided, it will be saved at the path provided with sodacl_yaml_path.

 required
Example 
from prefect_soda_core.sodacl_check import SodaCLCheck
sodacl_check_block = SodaCLCheck.load("BLOCK_NAME")
Source code in prefect_soda_core/sodacl_check.py 
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
class SodaCLCheck(Block):
    """
    This block represents a SodaCL check that can be used when running Soda scans.

    Args:
        sodacl_yaml_path (str): Absolute path of the Soda Checks file.
        sodacl_yaml_str (str): Optional YAML string containing the Soda Checks
            details. If provided, it will be saved
            at the path provided with `sodacl_yaml_path`.

    Example:
        ```python
        from prefect_soda_core.sodacl_check import SodaCLCheck
        sodacl_check_block = SodaCLCheck.load("BLOCK_NAME")
        ```
    """

    sodacl_yaml_path: str
    sodacl_yaml_str: Optional[str]

    _block_type_name: Optional[str] = "SodaCL Check"
    _logo_url: Optional[
        HttpUrl
    ] = "https://raw.githubusercontent.com/PrefectHQ/prefect/main/docs/img/collections/soda.png"  # noqa

    @root_validator(pre=True)
    def check_block_configuration(cls, values):
        """
        Ensure that the check configuration options are valid.
        A check configuration is valid if it provides just the path to the
        YAML Soda checks file or if it has both the path
        to the Soda checks file and a valid YAML Soda checks string.

        Raises:
            SodaConfigurationException: When the provided checks configuration
                is not valid.
        """
        sodacl_yaml_str_exists = bool(values.get("sodacl_yaml_str"))

        # If the YAML string is passed, but is not a valid YAML, then raise error
        if sodacl_yaml_str_exists:
            try:
                yaml_str = values.get("sodacl_yaml_str")
                safe_load(yaml_str)
            except YAMLError as exc:
                msg = f"The provided checks YAML is not valid. Error is: {exc}"
                raise SodaConfigurationException(msg)

        return values

    def persist_checks(self):
        """
        Persist Soda checks on the file system, if necessary.
        Please note that, if the path already exists, it will be overwritten.
        """

        # If a YAML string and path are passed, then persist the configuration
        if self.sodacl_yaml_str and self.sodacl_yaml_path:
            with open(self.sodacl_yaml_path, "w") as f:
                safe_dump(data=self.sodacl_yaml_str, stream=f)

check_block_configuration(values)

Ensure that the check configuration options are valid. A check configuration is valid if it provides just the path to the YAML Soda checks file or if it has both the path to the Soda checks file and a valid YAML Soda checks string.

Raises:

Type Description
SodaConfigurationException

When the provided checks configuration is not valid.
Source code in prefect_soda_core/sodacl_check.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@root_validator(pre=True)
def check_block_configuration(cls, values):
    """
    Ensure that the check configuration options are valid.
    A check configuration is valid if it provides just the path to the
    YAML Soda checks file or if it has both the path
    to the Soda checks file and a valid YAML Soda checks string.

    Raises:
        SodaConfigurationException: When the provided checks configuration
            is not valid.
    """
    sodacl_yaml_str_exists = bool(values.get("sodacl_yaml_str"))

    # If the YAML string is passed, but is not a valid YAML, then raise error
    if sodacl_yaml_str_exists:
        try:
            yaml_str = values.get("sodacl_yaml_str")
            safe_load(yaml_str)
        except YAMLError as exc:
            msg = f"The provided checks YAML is not valid. Error is: {exc}"
            raise SodaConfigurationException(msg)

    return values

persist_checks()

Persist Soda checks on the file system, if necessary. Please note that, if the path already exists, it will be overwritten.

Source code in prefect_soda_core/sodacl_check.py 
62
63
64
65
66
67
68
69
70
71
def persist_checks(self):
    """
    Persist Soda checks on the file system, if necessary.
    Please note that, if the path already exists, it will be overwritten.
    """

    # If a YAML string and path are passed, then persist the configuration
    if self.sodacl_yaml_str and self.sodacl_yaml_path:
        with open(self.sodacl_yaml_path, "w") as f:
            safe_dump(data=self.sodacl_yaml_str, stream=f)