configuration

drunc.utils.configuration

Configuration utilities for DRUNC.

Classes

ConfHandler

Handler for loading and parsing DRUNC configurations.

Supports multiple configuration sources via from_* classmethods. Subclasses override populate_from_dict / populate_from_pbany to handle JSON and protobuf sources, and _post_process_oks to handle OKS/pyobject sources (via self._raw_data).

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

populate_from_dict(data)

Populate from a dictionary (JSON source).

Override in subclasses that support JSON configuration.

Source code in drunc/utils/configuration.py

def populate_from_dict(self, data: dict[str, object]) -> None:
    """Populate from a dictionary (JSON source).

    Override in subclasses that support JSON configuration.
    """
    raise ConfTypeNotSupported(ConfTypes.JsonFileName, self.__class__.__name__)

populate_from_pbany(pbany_data)

Populate from a Protobuf Any message.

Override in subclasses that support protobuf configuration.

Source code in drunc/utils/configuration.py

def populate_from_pbany(self, pbany_data: object) -> None:
    """Populate from a Protobuf Any message.

    Override in subclasses that support protobuf configuration.
    """
    raise ConfTypeNotSupported(ConfTypes.ProtobufAny, 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