57 lines
2.1 KiB
Python
57 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from typing import Self, Any
|
|
from Interfaces.Application.NucelarMonitorInterface import NucelarMonitorInterface
|
|
from Abstracts.WebServerAbstract import WebServerAbstract
|
|
|
|
class WebServersManager:
|
|
|
|
def __init__(self:Self, nucelar_monitor:NucelarMonitorInterface) -> None:
|
|
self.nucelar_monitor:NucelarMonitorInterface = nucelar_monitor
|
|
|
|
key:str
|
|
|
|
self.__web_servers:dict[str, WebServerAbstract] = {}
|
|
|
|
for key in (
|
|
"default_web_servers_files", "web_servers_files",
|
|
"default_web_servers", "web_servers"
|
|
):
|
|
self.add(self.nucelar_monitor.settings.get(key, None, []), True)
|
|
|
|
def add(self:Self, inputs:Any|None, overwrite:bool = False) -> None:
|
|
|
|
subinputs:dict[str, WebServerAbstract|dict[str, Any|None]]
|
|
|
|
for subinputs in self.nucelar_monitor.files.load_json(inputs):
|
|
|
|
key:str
|
|
web_server:WebServerAbstract|dict[str, Any|None]
|
|
|
|
for key, web_server in subinputs.items():
|
|
if isinstance(web_server, dict):
|
|
web_server = self.nucelar_monitor.models.get(
|
|
WebServerAbstract,
|
|
self.nucelar_monitor.settings.get(("web_server_type", "type"), web_server, "web_server")
|
|
)(self.nucelar_monitor, web_server)
|
|
if web_server is not None and isinstance(web_server, WebServerAbstract) and (
|
|
overwrite or key not in self.__web_servers
|
|
):
|
|
self.__web_servers[key] = web_server
|
|
|
|
def start(self:Self) -> None:
|
|
|
|
web_server:WebServerAbstract
|
|
|
|
for web_server in self.__web_servers.values():
|
|
if hasattr(web_server, "start") and callable(getattr(web_server, "start")):
|
|
web_server.start()
|
|
|
|
def close(self:Self) -> None:
|
|
|
|
web_server:WebServerAbstract
|
|
|
|
for web_server in self.__web_servers.values():
|
|
if hasattr(web_server, "close") and callable(getattr(web_server, "close")):
|
|
web_server.close() |