Skip to content

Config utils

Module to read configurations.

ConfigUtils

Bases: object

Config utilities class.

Source code in mkdocs/lakehouse_engine/packages/utils/configs/config_utils.py 
 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
 76
 77
 78
 79
 80
 81
 82
 83
 84
 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
class ConfigUtils(object):
    """Config utilities class."""

    _LOGGER = LoggingHandler(__name__).get_logger()
    SENSITIVE_INFO = [
        "kafka.ssl.keystore.password",
        "kafka.ssl.truststore.password",
        "password",
        "secret",
        "credential",
        "credentials",
        "pass",
        "key",
    ]

    @classmethod
    def get_acon(
        cls,
        acon_path: Optional[str] = None,
        acon: Optional[dict] = None,
        disable_dbfs_retry: bool = False,
    ) -> dict:
        """Get acon based on a filesystem path or on a dict.

        Args:
            acon_path: path of the acon (algorithm configuration) file.
            acon: acon provided directly through python code (e.g., notebooks
                or other apps).
            disable_dbfs_retry: optional flag to disable file storage dbfs.

        Returns:
            Dict representation of an acon.
        """
        acon = (
            acon if acon else ConfigUtils.read_json_acon(acon_path, disable_dbfs_retry)
        )
        return acon

    @staticmethod
    def get_config(package: str = "lakehouse_engine.configs") -> Any:
        """Get the lakehouse engine configuration file.

        Returns:
            Configuration dictionary
        """
        with importlib.resources.open_binary(package, "engine.yaml") as config:
            config = yaml.safe_load(config)
        return config

    @staticmethod
    def get_config_from_file(config_file_path: str) -> Any:
        """Get the lakehouse engine configurations using a file path.

         Args:
            config_file_path: a string with a path for a yaml file
            with custom configurations.

        Returns:
            Configuration dictionary
        """
        with open(config_file_path, "r") as config:
            config = yaml.safe_load(config)
        return config

    @classmethod
    def get_engine_version(cls) -> str:
        """Get Lakehouse Engine version from the installed packages.

        Returns:
            String of engine version.
        """
        try:
            version = pkg_resources.get_distribution("lakehouse-engine").version
        except pkg_resources.DistributionNotFound:
            cls._LOGGER.info("Could not identify Lakehouse Engine version.")
            version = ""
        return str(version)

    @staticmethod
    def read_json_acon(path: str, disable_dbfs_retry: bool = False) -> Any:
        """Read an acon (algorithm configuration) file.

        Args:
            path: path to the acon file.
            disable_dbfs_retry: optional flag to disable file storage dbfs.

        Returns:
            The acon file content as a dict.
        """
        return FileStorageFunctions.read_json(path, disable_dbfs_retry)

    @staticmethod
    def read_sql(path: str, disable_dbfs_retry: bool = False) -> Any:
        """Read a DDL file in Spark SQL format from a cloud object storage system.

        Args:
            path: path to the SQL file.
            disable_dbfs_retry: optional flag to disable file storage dbfs.

        Returns:
            Content of the SQL file.
        """
        return FileStorageFunctions.read_sql(path, disable_dbfs_retry)

    @classmethod
    def remove_sensitive_info(
        cls, dict_to_replace: Union[dict, list]
    ) -> Union[dict, list]:
        """Remove sensitive info from a dictionary.

        Args:
            dict_to_replace: dict where we want to remove sensitive info.

        Returns:
            dict without sensitive information.
        """
        if isinstance(dict_to_replace, list):
            return [cls.remove_sensitive_info(k) for k in dict_to_replace]
        elif isinstance(dict_to_replace, dict):
            return {
                k: "******" if k in cls.SENSITIVE_INFO else cls.remove_sensitive_info(v)
                for k, v in dict_to_replace.items()
            }
        else:
            return dict_to_replace

get_acon(acon_path=None, acon=None, disable_dbfs_retry=False) classmethod

Get acon based on a filesystem path or on a dict.

Parameters:

Name Type Description Default
acon_path Optional[str]

path of the acon (algorithm configuration) file.

 None
acon Optional[dict]

acon provided directly through python code (e.g., notebooks or other apps).

 None
disable_dbfs_retry bool

optional flag to disable file storage dbfs.

 False

Returns:

Type Description
dict

Dict representation of an acon.
Source code in mkdocs/lakehouse_engine/packages/utils/configs/config_utils.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@classmethod
def get_acon(
    cls,
    acon_path: Optional[str] = None,
    acon: Optional[dict] = None,
    disable_dbfs_retry: bool = False,
) -> dict:
    """Get acon based on a filesystem path or on a dict.

    Args:
        acon_path: path of the acon (algorithm configuration) file.
        acon: acon provided directly through python code (e.g., notebooks
            or other apps).
        disable_dbfs_retry: optional flag to disable file storage dbfs.

    Returns:
        Dict representation of an acon.
    """
    acon = (
        acon if acon else ConfigUtils.read_json_acon(acon_path, disable_dbfs_retry)
    )
    return acon

get_config(package='lakehouse_engine.configs') staticmethod

Get the lakehouse engine configuration file.

Returns:

Type Description
Any

Configuration dictionary
Source code in mkdocs/lakehouse_engine/packages/utils/configs/config_utils.py 
51
52
53
54
55
56
57
58
59
60
@staticmethod
def get_config(package: str = "lakehouse_engine.configs") -> Any:
    """Get the lakehouse engine configuration file.

    Returns:
        Configuration dictionary
    """
    with importlib.resources.open_binary(package, "engine.yaml") as config:
        config = yaml.safe_load(config)
    return config

get_config_from_file(config_file_path) staticmethod

Get the lakehouse engine configurations using a file path.

Args: config_file_path: a string with a path for a yaml file with custom configurations.

Returns:

Type Description
Any

Configuration dictionary
Source code in mkdocs/lakehouse_engine/packages/utils/configs/config_utils.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@staticmethod
def get_config_from_file(config_file_path: str) -> Any:
    """Get the lakehouse engine configurations using a file path.

     Args:
        config_file_path: a string with a path for a yaml file
        with custom configurations.

    Returns:
        Configuration dictionary
    """
    with open(config_file_path, "r") as config:
        config = yaml.safe_load(config)
    return config

get_engine_version() classmethod

Get Lakehouse Engine version from the installed packages.

Returns:

Type Description
str

String of engine version.
Source code in mkdocs/lakehouse_engine/packages/utils/configs/config_utils.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
@classmethod
def get_engine_version(cls) -> str:
    """Get Lakehouse Engine version from the installed packages.

    Returns:
        String of engine version.
    """
    try:
        version = pkg_resources.get_distribution("lakehouse-engine").version
    except pkg_resources.DistributionNotFound:
        cls._LOGGER.info("Could not identify Lakehouse Engine version.")
        version = ""
    return str(version)

read_json_acon(path, disable_dbfs_retry=False) staticmethod

Read an acon (algorithm configuration) file.

Parameters:

Name Type Description Default
path str

path to the acon file.

 required
disable_dbfs_retry bool

optional flag to disable file storage dbfs.

 False

Returns:

Type Description
Any

The acon file content as a dict.
Source code in mkdocs/lakehouse_engine/packages/utils/configs/config_utils.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@staticmethod
def read_json_acon(path: str, disable_dbfs_retry: bool = False) -> Any:
    """Read an acon (algorithm configuration) file.

    Args:
        path: path to the acon file.
        disable_dbfs_retry: optional flag to disable file storage dbfs.

    Returns:
        The acon file content as a dict.
    """
    return FileStorageFunctions.read_json(path, disable_dbfs_retry)

read_sql(path, disable_dbfs_retry=False) staticmethod

Read a DDL file in Spark SQL format from a cloud object storage system.

Parameters:

Name Type Description Default
path str

path to the SQL file.

 required
disable_dbfs_retry bool

optional flag to disable file storage dbfs.

 False

Returns:

Type Description
Any

Content of the SQL file.
Source code in mkdocs/lakehouse_engine/packages/utils/configs/config_utils.py 
104
105
106
107
108
109
110
111
112
113
114
115
@staticmethod
def read_sql(path: str, disable_dbfs_retry: bool = False) -> Any:
    """Read a DDL file in Spark SQL format from a cloud object storage system.

    Args:
        path: path to the SQL file.
        disable_dbfs_retry: optional flag to disable file storage dbfs.

    Returns:
        Content of the SQL file.
    """
    return FileStorageFunctions.read_sql(path, disable_dbfs_retry)

remove_sensitive_info(dict_to_replace) classmethod

Remove sensitive info from a dictionary.

Parameters:

Name Type Description Default
dict_to_replace Union[dict, list]

dict where we want to remove sensitive info.

 required

Returns:

Type Description
Union[dict, list]

dict without sensitive information.
Source code in mkdocs/lakehouse_engine/packages/utils/configs/config_utils.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@classmethod
def remove_sensitive_info(
    cls, dict_to_replace: Union[dict, list]
) -> Union[dict, list]:
    """Remove sensitive info from a dictionary.

    Args:
        dict_to_replace: dict where we want to remove sensitive info.

    Returns:
        dict without sensitive information.
    """
    if isinstance(dict_to_replace, list):
        return [cls.remove_sensitive_info(k) for k in dict_to_replace]
    elif isinstance(dict_to_replace, dict):
        return {
            k: "******" if k in cls.SENSITIVE_INFO else cls.remove_sensitive_info(v)
            for k, v in dict_to_replace.items()
        }
    else:
        return dict_to_replace