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

entity_type_context

entity_type_context(model_spec: ModelSpec) -> Iterator[None]

Context manager for AttrSpec initialization.

Source code in dp3/common/config.py

@contextmanager
def entity_type_context(model_spec: ModelSpec) -> Iterator[None]:
    """Context manager for AttrSpec initialization."""
    token = _init_entity_type_context_var.set(
        {entity: spec.id_data_type.root for entity, spec in model_spec.entities.items()}
    )
    try:
        yield
    finally:
        _init_entity_type_context_var.reset(token)

get_entity_type_context

get_entity_type_context() -> dict

Get entity spec context.

Source code in dp3/common/config.py

def get_entity_type_context() -> dict:
    """Get entity spec context."""
    cxt = _init_entity_type_context_var.get()
    if cxt is None or not isinstance(cxt, dict):
        raise ValueError("Entity type context is not set")
    return cxt