configuration

drunc.utils.configuration

Configuration utilities for DRUNC.

Classes

ConfHandler(data=None, type=ConfTypes.PyObject, oks_key=None, *args, **kwargs)

Handler for loading and parsing DRUNC configurations.

Supports multiple configuration types including JSON files, Protobuf messages, and OKS.

Initialize a ConfHandler.

Parameters:

Name	Type	Description	Default
data	object	The configuration data. Defaults to None.	None
type	ConfTypes	The configuration type. Defaults to PyObject.	PyObject
oks_key	OKSKey | None	OKS key if using OKS configuration. Defaults to None.	None
*args	object	Additional positional arguments.	()
**kwargs	object	Additional keyword arguments.	{}

Raises:

Type	Description
DruncSetupException	If OKS type is used without an OKS key.

Source code in drunc/utils/configuration.py

def __init__(
    self,
    data: object = None,
    type: ConfTypes = ConfTypes.PyObject,
    oks_key: OKSKey | None = None,
    *args: object,
    **kwargs: object,
) -> None:
    """Initialize a ConfHandler.

    Args:
        data: The configuration data. Defaults to None.
        type: The configuration type. Defaults to PyObject.
        oks_key: OKS key if using OKS configuration. Defaults to None.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Raises:
        DruncSetupException: If OKS type is used without an OKS key.
    """
    self.class_name = self.__class__.__name__
    self.log = get_logger("utils." + self.class_name)
    self.initial_type = type
    self.initial_data = data
    self.root_id = 0
    self.controller_id = 0
    self.process_id = 0
    self.process_id_infra = 0
    self.session_name = kwargs.get("session_name")

    if type == ConfTypes.OKSFileName and oks_key is None:
        raise DruncSetupException("Need to provide a key for the OKS file")

    self.oks_key = oks_key
    self.validate_and_parse_configuration_location(*args, **kwargs)

Methods:

copy_oks_key()

Get a copy of the OKS key if one exists.

Returns:

Type	Description
OKSKey | None	OKSKey | None: The OKS key, or None if not using OKS configuration.

Source code in drunc/utils/configuration.py

def copy_oks_key(self) -> OKSKey | None:
    """Get a copy of the OKS key if one exists.

    Returns:
        OKSKey | None: The OKS key, or None if not using OKS configuration.
    """
    return self.oks_key

get_data()

Get the configuration data.

Returns:

Name	Type	Description
Any	object	The stored configuration data.

Source code in drunc/utils/configuration.py

def get_data(self) -> object:
    """Get the configuration data.

    Returns:
        Any: The stored configuration data.
    """
    return self.data

get_data_authoriser()

Get the authoriser from the configuration data.

Returns:

Name	Type	Description
Any	object	The authoriser object.

Source code in drunc/utils/configuration.py

def get_data_authoriser(self) -> object:
    """Get the authoriser from the configuration data.

    Returns:
        Any: The authoriser object.
    """
    return cast(_ConfigurationData, self.get_data()).authoriser

get_data_type_name()

Get the type name of the configuration data.

Returns:

Name	Type	Description
str	str	The name of the data type.

Source code in drunc/utils/configuration.py

def get_data_type_name(self) -> str:
    """Get the type name of the configuration data.

    Returns:
        str: The name of the data type.
    """
    return str(cast(_ConfigurationData, self.get_data()).type._name_)

validate_and_parse_configuration_location(*args, **kwargs)

Validate and parse the configuration from the provided location.

Supports JsonFileName, OKSFileName, and PyObject types.

Parameters:

Name	Type	Description	Default
*args	object	Additional positional arguments.	()
**kwargs	object	Additional keyword arguments.	{}

Source code in drunc/utils/configuration.py

def validate_and_parse_configuration_location(
    self, *args: object, **kwargs: object
) -> None:
    """Validate and parse the configuration from the provided location.

    Supports JsonFileName, OKSFileName, and PyObject types.

    Args:
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.
    """
    match self.initial_type:
        case ConfTypes.PyObject:
            self.data = self.initial_data
            self.type = self.initial_type
            self._post_process_oks(*args, **kwargs)

        case ConfTypes.JsonFileName:
            resolved = expand_path(cast(str, self.initial_data), True)
            if not os.path.exists(expand_path(cast(str, self.initial_data))):
                raise DruncSetupException(
                    f"Location {resolved} ({self.initial_data}) is empty!"
                )

            with open(resolved) as f:
                data = json.loads(f.read())
                self.data = self._parse_dict(data)
                self.type = ConfTypes.PyObject
                self._post_process_oks(*args, **kwargs)

        case ConfTypes.OKSFileName:
            self.data = self._parse_oks_file(cast(str, self.initial_data))
            self.type = ConfTypes.PyObject
            self._post_process_oks(*args, **kwargs)

        case ConfTypes.ProtobufAny:
            self.data = self._parse_pbany(self.initial_data)
            self.type = ConfTypes.PyObject
            self._post_process_oks(*args, **kwargs)

        case _:
            raise ConfTypeNotSupported(self.initial_type, self.class_name)

ConfTypeNotSupported(conf_type, class_name)

Bases: DruncSetupException

Exception raised when a configuration type is not supported.

Initialize the ConfTypeNotSupported exception.

Parameters:

Name	Type	Description	Default
conf_type	ConfTypes	The configuration type that is not supported.	required
class_name	str	The name of the class where this type is not supported.	required

Source code in drunc/utils/configuration.py

def __init__(self, conf_type: ConfTypes, class_name: str) -> None:
    """Initialize the ConfTypeNotSupported exception.

    Args:
        conf_type: The configuration type that is not supported.
        class_name: The name of the class where this type is not supported.
    """
    if not isinstance(class_name, str):
        class_name = class_name.__class__.__name__
    message = f"'{conf_type}' is not supported by '{class_name}'"
    super().__init__(message)

Methods:

ConfTypes

Bases: Enum

Enumeration of supported configuration types.

ConfigurationNotFound(requested_path)

Bases: DruncSetupException

Exception raised when configuration is not found.

Initialize the ConfigurationNotFound exception.

Parameters:

Name	Type	Description	Default
requested_path	str	The path to the configuration that was not found.	required

Source code in drunc/utils/configuration.py

def __init__(self, requested_path: str) -> None:
    """Initialize the ConfigurationNotFound exception.

    Args:
        requested_path: The path to the configuration that was not found.
    """
    super().__init__(
        f"The configuration '{requested_path}' is not in $DUNEDAQ_DB_PATH, perhaps you forgot to 'dbt-workarea-env && dbt-build'?"
    )

Methods:

OKSKey(schema_file, class_name, obj_uid, session)

Key information for accessing OKS configuration objects.

Initialize an OKSKey.

Parameters:

Name	Type	Description	Default
schema_file	str	The OKS schema file path.	required
class_name	str	The class name in the OKS schema.	required
obj_uid	str	The unique identifier for the object.	required
session	str	The session name.	required

Source code in drunc/utils/configuration.py

def __init__(
    self, schema_file: str, class_name: str, obj_uid: str, session: str
) -> None:
    """Initialize an OKSKey.

    Args:
        schema_file: The OKS schema file path.
        class_name: The class name in the OKS schema.
        obj_uid: The unique identifier for the object.
        session: The session name.
    """
    self.schema_file = schema_file
    self.class_name = class_name
    self.obj_uid = obj_uid
    self.session = session

Methods:

Functions:

CLI_to_ConfTypes(scheme)

Convert a CLI scheme string to a ConfTypes enum.

Parameters:

Name	Type	Description	Default
scheme	str	The scheme string ("file", "oksconflibs", or "").	required

Returns:

Name	Type	Description
ConfTypes	ConfTypes	The corresponding configuration type.

Raises:

Type	Description
DruncSetupException	If the scheme is not recognized.

Source code in drunc/utils/configuration.py

def CLI_to_ConfTypes(scheme: str) -> ConfTypes:
    """Convert a CLI scheme string to a ConfTypes enum.

    Args:
        scheme: The scheme string ("file", "oksconflibs", or "").

    Returns:
        ConfTypes: The corresponding configuration type.

    Raises:
        DruncSetupException: If the scheme is not recognized.
    """
    match scheme:
        case "file":
            return ConfTypes.JsonFileName
        case "oksconflibs" | "":
            return ConfTypes.OKSFileName
        case _:
            raise DruncSetupException(f"{scheme} configuration type is not understood")

parse_conf_url(url)

Parse a configuration URL into scheme and type.

Parameters:

Name	Type	Description	Default
url	str	The configuration URL (format: "scheme:filename").	required

Returns:

Type	Description
tuple[str, ConfTypes]	tuple[str, ConfTypes]: A tuple of (url, conf_type).

Source code in drunc/utils/configuration.py

def parse_conf_url(url: str) -> tuple[str, ConfTypes]:
    """Parse a configuration URL into scheme and type.

    Args:
        url: The configuration URL (format: "scheme:filename").

    Returns:
        tuple[str, ConfTypes]: A tuple of (url, conf_type).
    """
    scheme, filename = url.split(":")
    t = CLI_to_ConfTypes(scheme)
    return url, t