afonso/core-extra
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

daemon: updating core.configservice and core.configservices to avoid deprecated type hinting

Browse source
This commit is contained in:
Blake Harnden 2023-04-13 11:58:58 -07:00
parent e770bcd47c
commit 3d722a7721
8 changed files with 345 additions and 346 deletions
Download patch file Download diff file Expand all files Collapse all files

48
daemon/core/configservice/base.py
View file

@ -5,7 +5,7 @@ import logging
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional
from typing import Any, Optional
from mako import exceptions
from mako.lookup import TemplateLookup
@ -67,7 +67,7 @@ class ConfigService(abc.ABC):
validation_timer: int = 5
# directories to shadow and copy files from
shadow_directories: List[ShadowDir] = []
shadow_directories: list[ShadowDir] = []
def __init__(self, node: CoreNode) -> None:
"""
@ -79,9 +79,9 @@ class ConfigService(abc.ABC):
class_file = inspect.getfile(self.__class__)
templates_path = Path(class_file).parent.joinpath(TEMPLATES_DIR)
self.templates: TemplateLookup = TemplateLookup(directories=templates_path)
self.config: Dict[str, Configuration] = {}
self.custom_templates: Dict[str, str] = {}
self.custom_config: Dict[str, str] = {}
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)
@ -108,47 +108,47 @@ class ConfigService(abc.ABC):
@property
@abc.abstractmethod
def directories(self) -> List[str]:
def directories(self) -> list[str]:
raise NotImplementedError
@property
@abc.abstractmethod
def files(self) -> List[str]:
def files(self) -> list[str]:
raise NotImplementedError
@property
@abc.abstractmethod
def default_configs(self) -> List[Configuration]:
def default_configs(self) -> list[Configuration]:
raise NotImplementedError
@property
@abc.abstractmethod
def modes(self) -> Dict[str, Dict[str, str]]:
def modes(self) -> dict[str, dict[str, str]]:
raise NotImplementedError
@property
@abc.abstractmethod
def executables(self) -> List[str]:
def executables(self) -> list[str]:
raise NotImplementedError
@property
@abc.abstractmethod
def dependencies(self) -> List[str]:
def dependencies(self) -> list[str]:
raise NotImplementedError
@property
@abc.abstractmethod
def startup(self) -> List[str]:
def startup(self) -> list[str]:
raise NotImplementedError
@property
@abc.abstractmethod
def validate(self) -> List[str]:
def validate(self) -> list[str]:
raise NotImplementedError
@property
@abc.abstractmethod
def shutdown(self) -> List[str]:
def shutdown(self) -> list[str]:
raise NotImplementedError
@property
@ -276,7 +276,7 @@ class ConfigService(abc.ABC):
f"failure to create service directory: {directory}"
)
def data(self) -> Dict[str, Any]:
def data(self) -> dict[str, Any]:
"""
Returns key/value data, used when rendering file templates.
@ -303,7 +303,7 @@ class ConfigService(abc.ABC):
"""
raise CoreError(f"service({self.name}) unknown template({name})")
def get_templates(self) -> Dict[str, str]:
def get_templates(self) -> dict[str, str]:
"""
Retrieves mapping of file names to templates for all cases, which
includes custom templates, file templates, and text templates.
@ -331,7 +331,7 @@ class ConfigService(abc.ABC):
templates[file] = template
return templates
def get_rendered_templates(self) -> Dict[str, str]:
def get_rendered_templates(self) -> dict[str, str]:
templates = {}
data = self.data()
for file in sorted(self.files):
@ -339,7 +339,7 @@ class ConfigService(abc.ABC):
templates[file] = rendered
return templates
def _get_rendered_template(self, file: str, data: Dict[str, Any]) -> str:
def _get_rendered_template(self, file: str, data: dict[str, Any]) -> str:
file_path = Path(file)
template_path = get_template_path(file_path)
if file in self.custom_templates:
@ -426,7 +426,7 @@ class ConfigService(abc.ABC):
f"node({self.node.name}) service({self.name}) failed to validate"
)
def _render(self, template: Template, data: Dict[str, Any] = None) -> str:
def _render(self, template: Template, data: dict[str, Any] = None) -> str:
"""
Renders template providing all associated data to template.
@ -440,7 +440,7 @@ class ConfigService(abc.ABC):
node=self.node, config=self.render_config(), **data
)
def render_text(self, text: str, data: Dict[str, Any] = None) -> str:
def render_text(self, text: str, data: dict[str, Any] = None) -> str:
"""
Renders text based template providing all associated data to template.
@ -458,7 +458,7 @@ class ConfigService(abc.ABC):
f"{exceptions.text_error_template().render_unicode()}"
)
def render_template(self, template_path: str, data: Dict[str, Any] = None) -> str:
def render_template(self, template_path: str, data: dict[str, Any] = None) -> str:
"""
Renders file based template providing all associated data to template.
@ -475,7 +475,7 @@ class ConfigService(abc.ABC):
f"{exceptions.text_error_template().render_unicode()}"
)
def _define_config(self, configs: List[Configuration]) -> None:
def _define_config(self, configs: list[Configuration]) -> None:
"""
Initializes default configuration data.
@ -485,7 +485,7 @@ class ConfigService(abc.ABC):
for config in configs:
self.config[config.id] = config
def render_config(self) -> Dict[str, str]:
def render_config(self) -> dict[str, str]:
"""
Returns configuration data key/value pairs for rendering a template.
@ -496,7 +496,7 @@ class ConfigService(abc.ABC):
else:
return {k: v.default for k, v in self.config.items()}
def set_config(self, data: Dict[str, str]) -> None:
def set_config(self, data: dict[str, str]) -> None:
"""
Set configuration data from key/value pairs.

22
daemon/core/configservice/dependencies.py
View file

@ -1,5 +1,5 @@
import logging
from typing import TYPE_CHECKING, Dict, List, Set
from typing import TYPE_CHECKING
logger = logging.getLogger(__name__)
@ -12,16 +12,16 @@ class ConfigServiceDependencies:
Generates sets of services to start in order of their dependencies.
"""
def __init__(self, services: Dict[str, "ConfigService"]) -> None:
def __init__(self, services: dict[str, "ConfigService"]) -> None:
"""
Create a ConfigServiceDependencies instance.
:param services: services for determining dependency sets
"""
# helpers to check validity
self.dependents: Dict[str, Set[str]] = {}
self.started: Set[str] = set()
self.node_services: Dict[str, "ConfigService"] = {}
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:
@ -29,11 +29,11 @@ class ConfigServiceDependencies:
dependents.add(service.name)
# used to find paths
self.path: List["ConfigService"] = []
self.visited: Set[str] = set()
self.visiting: Set[str] = set()
self.path: list["ConfigService"] = []
self.visited: set[str] = set()
self.visiting: set[str] = set()
def startup_paths(self) -> List[List["ConfigService"]]:
def startup_paths(self) -> list[list["ConfigService"]]:
"""
Find startup path sets based on service dependencies.
@ -70,7 +70,7 @@ class ConfigServiceDependencies:
self.visited.clear()
self.visiting.clear()
def _start(self, service: "ConfigService") -> List["ConfigService"]:
def _start(self, service: "ConfigService") -> list["ConfigService"]:
"""
Starts a oath for checking dependencies for a given service.
@ -81,7 +81,7 @@ class ConfigServiceDependencies:
self._reset()
return self._visit(service)
def _visit(self, current_service: "ConfigService") -> List["ConfigService"]:
def _visit(self, current_service: "ConfigService") -> list["ConfigService"]:
"""
Visits a service when discovering dependency chains for service.

11
daemon/core/configservice/manager.py
View file

@ -2,7 +2,6 @@ import logging
import pathlib
import pkgutil
from pathlib import Path
from typing import Dict, List, Type
from core import configservices, utils
from core.configservice.base import ConfigService
@ -20,9 +19,9 @@ class ConfigServiceManager:
"""
Create a ConfigServiceManager instance.
"""
self.services: Dict[str, Type[ConfigService]] = {}
self.services: dict[str, type[ConfigService]] = {}
def get_service(self, name: str) -> Type[ConfigService]:
def get_service(self, name: str) -> type[ConfigService]:
"""
Retrieve a service by name.
@ -35,7 +34,7 @@ class ConfigServiceManager:
raise CoreError(f"service does not exist {name}")
return service_class
def add(self, service: Type[ConfigService]) -> None:
def add(self, service: type[ConfigService]) -> None:
"""
Add service to manager, checking service requirements have been met.
@ -62,7 +61,7 @@ class ConfigServiceManager:
# make service available
self.services[name] = service
def load_locals(self) -> List[str]:
def load_locals(self) -> list[str]:
"""
Search and add config service from local core module.
@ -81,7 +80,7 @@ class ConfigServiceManager:
logger.debug("not loading config service(%s): %s", service.name, e)
return errors
def load(self, path: Path) -> List[str]:
def load(self, path: Path) -> list[str]:
"""
Search path provided for config services and add them for being managed.