Source code for boss.configuration

import json

from .factory import DatasourceHEROFactory, HEROFactory, PolledDatasourceHEROFactory
from .helper import file_from_url, get_remote_hero, log




def _resolve_references(value: object, realm: str) -> object:
    """
    Replace ``$name`` strings at any depth with remote HEROs.

    A leading ``$$`` is an escape: one ``$`` is stripped and the string is passed on literally. Dict keys are not resolved.
    Containers are rebuilt, the input is not modified.

    Args:
        value: argument value, possibly a nested dict, list or tuple
        realm: realm in which the referenced HEROs are looked up

    Returns:
        The value with all references resolved.
    """
    if isinstance(value, str):
        if value.startswith("$$"):
            return value[1:]
        if value.startswith("$"):
            return get_remote_hero(value[1:], realm=realm)
        return value
    if isinstance(value, dict):
        return {key: _resolve_references(val, realm) for key, val in value.items()}
    if isinstance(value, list):
        return [_resolve_references(item, realm) for item in value]
    if isinstance(value, tuple):
        return tuple(_resolve_references(item, realm) for item in value)
    return value





class Document(dict):


    @classmethod
    def parse_string(cls, s: str):
        """
        Parse a document in JSON format from a string.

        Args:
            s: string containing the JSON representation of the config
        """
        try:
            obj = cls()
            obj.update(json.loads(s))
            return obj
        except json.JSONDecodeError as e:
            log.error(f"Error while encoding json: {e}")




    @classmethod
    def parse_url(cls, url: str):
        """
        Parse a document in JSON format from a URL.

        Args:
            url: any URL supported by urllib (e.g. file://local.json or https://user:pass@couch.db/database/my_doc)
        """
        f_handle = file_from_url(url)
        return cls.parse_string(f_handle.read())






class WorkerConfigurationDocument(Document):


    def datasource_config(self):
        if "datasource" in self and isinstance(self["datasource"], dict):
            cfg = {"async": False, "interval": 5.0, "observables": {}}
            cfg.update(self["datasource"])
            return cfg
        else:
            return None




    def generate_tags(self, boss_name):
        """Add boss/configuration specific tags to the HERO."""
        if "tags" not in self:
            self["tags"] = []
        self["tags"].append(f"BOSS: {boss_name}")




    def build_hero_for_worker(self, boss_name: str, worker_object, realm="heros"):
        # replace special string for asyncio loop and multiprocess pool
        self.generate_tags(boss_name)

        arguments = {
            key: worker_object._loop if val == "@_boss_loop" else val for key, val in self["arguments"].items()
        }
        self["arguments"] = _resolve_references(arguments, realm)

        base_config = {
            "classname": self["classname"],
            "arg_dict": self.get("arguments", {}),
            "extra_decorators": self.get("extra_decorators", {}),
            "wrap_target": self.get("wrap_target", None),
            "name": self["_id"],
            "realm": realm,
            "tags": self["tags"],
        }

        if (datasource_config := self.datasource_config()) is not None:
            if datasource_config["async"]:
                return DatasourceHEROFactory.build(
                    **base_config,
                    observables=datasource_config["observables"],
                )
            else:
                return PolledDatasourceHEROFactory.build(
                    **base_config,
                    loop=worker_object._loop,
                    interval=datasource_config["interval"],
                    observables=datasource_config["observables"],
                )
        else:
            return HEROFactory.build(**base_config)