daemon: added type hinting throughout config services
This commit is contained in:
parent
a09910d0bc
commit
b2ea8cbbf6
9 changed files with 471 additions and 455 deletions
|
@ -14,7 +14,7 @@ from core.config import Configuration
|
|||
from core.errors import CoreCommandError, CoreError
|
||||
from core.nodes.base import CoreNode
|
||||
|
||||
TEMPLATES_DIR = "templates"
|
||||
TEMPLATES_DIR: str = "templates"
|
||||
|
||||
|
||||
class ConfigServiceMode(enum.Enum):
|
||||
|
@ -33,10 +33,10 @@ class ConfigService(abc.ABC):
|
|||
"""
|
||||
|
||||
# validation period in seconds, how frequent validation is attempted
|
||||
validation_period = 0.5
|
||||
validation_period: float = 0.5
|
||||
|
||||
# time to wait in seconds for determining if service started successfully
|
||||
validation_timer = 5
|
||||
validation_timer: int = 5
|
||||
|
||||
def __init__(self, node: CoreNode) -> None:
|
||||
"""
|
||||
|
@ -44,13 +44,13 @@ class ConfigService(abc.ABC):
|
|||
|
||||
:param node: node this service is assigned to
|
||||
"""
|
||||
self.node = node
|
||||
self.node: CoreNode = node
|
||||
class_file = inspect.getfile(self.__class__)
|
||||
templates_path = pathlib.Path(class_file).parent.joinpath(TEMPLATES_DIR)
|
||||
self.templates = TemplateLookup(directories=templates_path)
|
||||
self.config = {}
|
||||
self.custom_templates = {}
|
||||
self.custom_config = {}
|
||||
self.templates: TemplateLookup = TemplateLookup(directories=templates_path)
|
||||
self.config: Dict[str, Configuration] = {}
|
||||
self.custom_templates: Dict[str, str] = {}
|
||||
self.custom_config: Dict[str, str] = {}
|
||||
configs = self.default_configs[:]
|
||||
self._define_config(configs)
|
||||
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
import logging
|
||||
from typing import TYPE_CHECKING, Dict, List
|
||||
from typing import TYPE_CHECKING, Dict, List, Set
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from core.configservice.base import ConfigService
|
||||
|
@ -17,9 +17,9 @@ class ConfigServiceDependencies:
|
|||
:param services: services for determining dependency sets
|
||||
"""
|
||||
# helpers to check validity
|
||||
self.dependents = {}
|
||||
self.started = set()
|
||||
self.node_services = {}
|
||||
self.dependents: Dict[str, Set[str]] = {}
|
||||
self.started: Set[str] = set()
|
||||
self.node_services: Dict[str, "ConfigService"] = {}
|
||||
for service in services.values():
|
||||
self.node_services[service.name] = service
|
||||
for dependency in service.dependencies:
|
||||
|
@ -27,9 +27,9 @@ class ConfigServiceDependencies:
|
|||
dependents.add(service.name)
|
||||
|
||||
# used to find paths
|
||||
self.path = []
|
||||
self.visited = set()
|
||||
self.visiting = set()
|
||||
self.path: List["ConfigService"] = []
|
||||
self.visited: Set[str] = set()
|
||||
self.visiting: Set[str] = set()
|
||||
|
||||
def startup_paths(self) -> List[List["ConfigService"]]:
|
||||
"""
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
import logging
|
||||
import pathlib
|
||||
from typing import List, Type
|
||||
from typing import Dict, List, Type
|
||||
|
||||
from core import utils
|
||||
from core.configservice.base import ConfigService
|
||||
|
@ -16,7 +16,7 @@ class ConfigServiceManager:
|
|||
"""
|
||||
Create a ConfigServiceManager instance.
|
||||
"""
|
||||
self.services = {}
|
||||
self.services: Dict[str, Type[ConfigService]] = {}
|
||||
|
||||
def get_service(self, name: str) -> Type[ConfigService]:
|
||||
"""
|
||||
|
@ -31,7 +31,7 @@ class ConfigServiceManager:
|
|||
raise CoreError(f"service does not exit {name}")
|
||||
return service_class
|
||||
|
||||
def add(self, service: ConfigService) -> None:
|
||||
def add(self, service: Type[ConfigService]) -> None:
|
||||
"""
|
||||
Add service to manager, checking service requirements have been met.
|
||||
|
||||
|
@ -40,7 +40,9 @@ class ConfigServiceManager:
|
|||
:raises CoreError: when service is a duplicate or has unmet executables
|
||||
"""
|
||||
name = service.name
|
||||
logging.debug("loading service: class(%s) name(%s)", service.__class__, name)
|
||||
logging.debug(
|
||||
"loading service: class(%s) name(%s)", service.__class__.__name__, name
|
||||
)
|
||||
|
||||
# avoid duplicate services
|
||||
if name in self.services:
|
||||
|
@ -73,7 +75,6 @@ class ConfigServiceManager:
|
|||
logging.debug("loading config services from: %s", subdir)
|
||||
services = utils.load_classes(str(subdir), ConfigService)
|
||||
for service in services:
|
||||
logging.debug("found service: %s", service)
|
||||
try:
|
||||
self.add(service)
|
||||
except CoreError as e:
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue