dp3.common.config ¶

Platform config file reader and config model.

HierarchicalDict ¶

Bases: dict

Extension of built-in dict that simplifies working with a nested hierarchy of dicts.

get ¶

get(key, default=NoDefault)

Key may be a path (in dot notation) into a hierarchy of dicts. For example dictionary.get('abc.x.y') is equivalent to dictionary['abc']['x']['y'].

:returns: self[key] or default if key is not found.

Source code in dp3/common/config.py

def get(self, key, default=NoDefault):
    """
    Key may be a path (in dot notation) into a hierarchy of dicts. For example
      `dictionary.get('abc.x.y')`
    is equivalent to
      `dictionary['abc']['x']['y']`.

    :returns: `self[key]` or `default` if key is not found.
    """
    d = self
    try:
        while "." in key:
            first_key, key = key.split(".", 1)
            d = d[first_key]
        return d[key]
    except (KeyError, TypeError):
        pass  # not found - continue below
    if default is NoDefault:
        raise MissingConfigError("Mandatory configuration element is missing: " + key)
    else:
        return default

update ¶

update(other, **kwargs)

Update HierarchicalDict with other dictionary and merge common keys.

If there is a key in both current and the other dictionary and values of both keys are dictionaries, they are merged together.

Example:

HierarchicalDict({'a': {'b': 1, 'c': 2}}).update({'a': {'b': 10, 'd': 3}})
->
HierarchicalDict({'a': {'b': 10, 'c': 2, 'd': 3}})

Changes the dictionary directly, returns None.

Source code in dp3/common/config.py

def update(self, other, **kwargs):
    """
    Update `HierarchicalDict` with other dictionary and merge common keys.

    If there is a key in both current and the other dictionary and values of
    both keys are dictionaries, they are merged together.

    Example:
    ```
    HierarchicalDict({'a': {'b': 1, 'c': 2}}).update({'a': {'b': 10, 'd': 3}})
    ->
    HierarchicalDict({'a': {'b': 10, 'c': 2, 'd': 3}})
    ```
    Changes the dictionary directly, returns `None`.
    """
    other = dict(other)
    for key in other:
        if key in self:
            if isinstance(self[key], dict) and isinstance(other[key], dict):
                # The key is present in both dicts and both key values are dicts -> merge them
                HierarchicalDict.update(self[key], other[key])
            else:
                # One of the key values is not a dict -> overwrite the value
                # in self by the one from other (like normal "update" does)
                self[key] = other[key]
        else:
            # key is not present in self -> set it to value from other
            self[key] = other[key]

CronExpression ¶

Bases: BaseModel

Cron expression used for scheduling. Also support standard cron expressions, such as

"*/15" (every 15 units)
"1,2,3" (1, 2 and 3)
"1-3" (1, 2 and 3)

Attributes:

Name	Type	Description
`year`	`Optional[str]`	4-digit year
`month`	`Optional[int]`	month (1-12)
`day`	`Optional[Union[Annotated[int, Field(ge=1, le=31)], CronStr]]`	day of month (1-31)
`week`	`Optional[int]`	ISO week (1-53)
`day_of_week`	`Optional[Union[Annotated[int, Field(ge=0, le=6)], CronStr]]`	number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)
`hour`	`Optional[Union[TimeInt, CronStr]]`	hour (0-23)
`minute`	`Optional[Union[TimeInt, CronStr]]`	minute (0-59)
`second`	`Optional[Union[TimeInt, CronStr]]`	second (0-59)
`timezone`	`str`	Timezone for time specification (default is UTC).

EntitySpecDict ¶

Bases: BaseModel

Class representing full specification of an entity.

Attributes:

Name	Type	Description
`entity`	`EntitySpec`	Specification and settings of entity itself.
`attribs`	`dict[str, AttrSpecType]`	A mapping of attribute id -> AttrSpec

ModelSpec ¶

ModelSpec(config: HierarchicalDict)

Bases: BaseModel

Class representing the platform's current entity and attribute specification.

Attributes:

Name	Type	Description
`config`	`dict[str, EntitySpecDict]`	Legacy config format, exactly mirrors the config files.
`entities`	`dict[str, EntitySpec]`	Mapping of entity id -> EntitySpec
`attributes`	`dict[tuple[str, str], AttrSpecType]`	Mapping of (entity id, attribute id) -> AttrSpec
`entity_attributes`	`dict[str, dict[str, AttrSpecType]]`	Mapping of entity id -> attribute id -> AttrSpec
`relations`	`dict[tuple[str, str], AttrSpecType]`	Mapping of (entity id, attribute id) -> AttrSpec only contains attributes which are relations.

Provided configuration must be a dict of following structure:

{
    <entity type>: {
        'entity': {
            entity specification
        },
        'attribs': {
            <attr id>: {
                attribute specification
            },
            other attributes
        }
    },
    other entity types
}

Raises: ValueError: if the specification is invalid.

Source code in dp3/common/config.py

def __init__(self, config: HierarchicalDict):
    """
    Provided configuration must be a dict of following structure:
    ```
    {
        <entity type>: {
            'entity': {
                entity specification
            },
            'attribs': {
                <attr id>: {
                    attribute specification
                },
                other attributes
            }
        },
        other entity types
    }
    ```
    Raises:
        ValueError: if the specification is invalid.
    """
    super().__init__(
        config=config, entities={}, attributes={}, entity_attributes={}, relations={}
    )

PlatformConfig ¶

Bases: BaseModel

An aggregation of configuration available to modules.

Attributes:

Name	Type	Description
`app_name`	`str`	Name of the application, used when naming various structures of the platform
`config_base_path`	`str`	Path to directory containing platform config
`config`	`HierarchicalDict`	A dictionary that contains the platform config
`model_spec`	`ModelSpec`	Specification of the platform's model (entities and attributes)
`num_processes`	`PositiveInt`	Number of worker processes
`process_index`	`NonNegativeInt`	Index of current process

read_config ¶

read_config(filepath: str) -> HierarchicalDict

Read configuration file and return config as a dict-like object.

The configuration file should contain a valid YAML - Comments may be included as lines starting with # (optionally preceded by whitespaces).

This function reads the file and converts it to a HierarchicalDict. The only difference from built-in dict is its get method, which allows hierarchical keys (e.g. abc.x.y). See doc of get method for more information.

Source code in dp3/common/config.py

def read_config(filepath: str) -> HierarchicalDict:
    """
    Read configuration file and return config as a dict-like object.

    The configuration file should contain a valid YAML
    - Comments may be included as lines starting with `#` (optionally preceded
      by whitespaces).

    This function reads the file and converts it to a `HierarchicalDict`.
    The only difference from built-in `dict` is its `get` method, which allows
    hierarchical keys (e.g. `abc.x.y`).
    See [doc of get method][dp3.common.config.HierarchicalDict.get] for more information.
    """
    with open(filepath) as file_content:
        return HierarchicalDict(yaml.safe_load(file_content))

read_config_dir ¶

read_config_dir(dir_path: str, recursive: bool = False) -> HierarchicalDict

Same as read_config, but it loads whole configuration directory of YAML files, so only files ending with ".yml" are loaded. Each loaded configuration is located under key named after configuration filename.

Parameters:

Name	Type	Description	Default
`dir_path`	`str`	Path to read config from.	required
`recursive`	`bool`	If `recursive` is set, then the configuration directory will be read recursively (including configuration files inside directories).	`False`

Source code in dp3/common/config.py

def read_config_dir(dir_path: str, recursive: bool = False) -> HierarchicalDict:
    """
    Same as [read_config][dp3.common.config.read_config],
    but it loads whole configuration directory of YAML files,
    so only files ending with ".yml" are loaded.
    Each loaded configuration is located under key named after configuration filename.

    Args:
        dir_path: Path to read config from.
        recursive: If `recursive` is set, then the configuration directory will be read
            recursively (including configuration files inside directories).
    """
    all_files_paths = os.listdir(dir_path)
    config = HierarchicalDict()
    for config_filename in all_files_paths:
        config_full_path = os.path.join(dir_path, config_filename)
        if os.path.isdir(config_full_path) and recursive:
            loaded_config = read_config_dir(config_full_path, recursive)
        elif os.path.isfile(config_full_path) and config_filename.endswith(".yml"):
            try:
                loaded_config = read_config(config_full_path)
            except TypeError:
                # configuration file is empty
                continue
            # remove '.yml' suffix of filename
            config_filename = config_filename[:-4]
        else:
            continue
        # place configuration files into another dictionary level named by config dictionary name
        config[config_filename] = loaded_config
    return config