core-extra/daemon/core/configservice/manager.py

import logging
import pathlib
from typing import List, Type

from core import utils
from core.configservice.base import ConfigService
from core.errors import CoreError


class ConfigServiceManager:
    def __init__(self):
        self.services = {}

    def get_service(self, name: str) -> Type[ConfigService]:
        service_class = self.services.get(name)
        if service_class is None:
            raise CoreError(f"service does not exit {name}")
        return service_class

    def add(self, service: ConfigService) -> None:
        name = service.name
        logging.debug("loading service: class(%s) name(%s)", service.__class__, name)

        # avoid duplicate services
        if name in self.services:
            raise CoreError(f"duplicate service being added: {name}")

        # validate dependent executables are present
        for executable in service.executables:
            try:
                utils.which(executable, required=True)
            except ValueError:
                raise CoreError(
                    f"service({service.name}) missing executable {executable}"
                )

            # make service available
        self.services[name] = service

    def load(self, path: str) -> List[str]:
        path = pathlib.Path(path)
        subdirs = [x for x in path.iterdir() if x.is_dir()]
        subdirs.append(path)
        service_errors = []
        for subdir in subdirs:
            logging.info("loading config services from: %s", subdir)
            services = utils.load_classes(str(subdir), ConfigService)
            for service in services:
                logging.info("found service: %s", service)
                try:
                    self.add(service)
                except CoreError as e:
                    service_errors.append(service.name)
                    logging.debug("not loading service(%s): %s", service.name, e)
        return service_errors