orrisroot
b95fc0cd7d
Abstract the configuration storage mechanism to allow using custom configurations, such as in-memory setups, when using the tool as a library. This aligns the configuration architecture with the session cache abstraction. - Define ConfigInterface protocol and InMemoryConfig class - Make CacheFile, InMemoryCache, ConfigFile, and InMemoryConfig explicitly inherit their interfaces - Update MdrsService and MdrsClient to accept customizable config_class and config instances - Add validation to check remote parameter consistency in create_connection - Remove unused imports across command files
134 lines
3.9 KiB
Python
134 lines
3.9 KiB
Python
import configparser
|
|
import os
|
|
import threading
|
|
from typing import Final, Protocol, runtime_checkable
|
|
|
|
import validators
|
|
|
|
from mdrsclient.exceptions import IllegalArgumentException
|
|
from mdrsclient.settings import CONFIG_DIRNAME
|
|
from mdrsclient.utils import FileLock
|
|
|
|
|
|
@runtime_checkable
|
|
class ConfigInterface(Protocol):
|
|
remote: str
|
|
|
|
def list(self) -> list[tuple[str, str]]: ...
|
|
@property
|
|
def url(self) -> str | None: ...
|
|
@url.setter
|
|
def url(self, url: str) -> None: ...
|
|
@url.deleter
|
|
def url(self) -> None: ...
|
|
|
|
|
|
class InMemoryConfig(ConfigInterface):
|
|
__configs: dict[str, str] = {}
|
|
__lock: threading.Lock = threading.Lock()
|
|
remote: str
|
|
|
|
def __init__(self, remote: str) -> None:
|
|
self.remote = remote
|
|
|
|
def list(self) -> list[tuple[str, str]]:
|
|
with self.__lock:
|
|
return list(self.__configs.items())
|
|
|
|
@property
|
|
def url(self) -> str | None:
|
|
with self.__lock:
|
|
return self.__configs.get(self.remote)
|
|
|
|
@url.setter
|
|
def url(self, url: str) -> None:
|
|
if not validators.url(url):
|
|
raise IllegalArgumentException("malformed URI sequence")
|
|
with self.__lock:
|
|
self.__configs[self.remote] = url
|
|
|
|
@url.deleter
|
|
def url(self) -> None:
|
|
with self.__lock:
|
|
if self.remote in self.__configs:
|
|
del self.__configs[self.remote]
|
|
|
|
@classmethod
|
|
def clear(cls) -> None:
|
|
with cls.__lock:
|
|
cls.__configs.clear()
|
|
|
|
|
|
class ConfigFile(ConfigInterface):
|
|
OPTION_URL: Final[str] = "url"
|
|
CONFIG_FILENAME: Final[str] = "config.ini"
|
|
remote: str
|
|
__serial: int
|
|
__config_dirname: str
|
|
__config_path: str
|
|
__config: configparser.ConfigParser
|
|
|
|
def __init__(self, remote: str) -> None:
|
|
self.remote = remote
|
|
self.__serial = -1
|
|
self.__config_dirname = CONFIG_DIRNAME
|
|
self.__config_path = os.path.join(CONFIG_DIRNAME, self.CONFIG_FILENAME)
|
|
self.__config = configparser.ConfigParser()
|
|
|
|
def list(self) -> list[tuple[str, str]]:
|
|
ret: list[tuple[str, str]] = []
|
|
self.__load()
|
|
for remote in self.__config.sections():
|
|
url = self.__config.get(remote, self.OPTION_URL)
|
|
ret.append((remote, url))
|
|
return ret
|
|
|
|
@property
|
|
def url(self) -> str | None:
|
|
if not self.__exists(self.remote):
|
|
return None
|
|
return self.__config.get(self.remote, self.OPTION_URL)
|
|
|
|
@url.setter
|
|
def url(self, url: str) -> None:
|
|
if not validators.url(url):
|
|
raise IllegalArgumentException("malformed URI sequence")
|
|
self.__load()
|
|
if self.__config.has_section(self.remote):
|
|
self.__config.remove_section(self.remote)
|
|
self.__config.add_section(self.remote)
|
|
self.__config.set(self.remote, self.OPTION_URL, url)
|
|
self.__save()
|
|
|
|
@url.deleter
|
|
def url(self) -> None:
|
|
if self.__exists(self.remote):
|
|
self.__config.remove_section(self.remote)
|
|
self.__save()
|
|
|
|
def __exists(self, section: str) -> bool:
|
|
self.__load()
|
|
return self.__config.has_option(section, self.OPTION_URL)
|
|
|
|
def __load(self) -> None:
|
|
if os.path.isfile(self.__config_path):
|
|
stat = os.stat(self.__config_path)
|
|
serial = hash(stat)
|
|
if self.__serial != serial:
|
|
self.__config.read(self.__config_path, encoding="utf8")
|
|
self.__serial = serial
|
|
|
|
def __save(self) -> None:
|
|
self.__ensure_cache_dir()
|
|
with open(self.__config_path, "w") as f:
|
|
FileLock.lock(f)
|
|
self.__config.write(f)
|
|
FileLock.unlock(f)
|
|
os.chmod(self.__config_path, 0o600)
|
|
|
|
def __ensure_cache_dir(self) -> None:
|
|
if not os.path.exists(self.__config_dirname):
|
|
os.makedirs(self.__config_dirname)
|
|
# ensure directory is secure.
|
|
os.chmod(self.__config_dirname, 0o700)
|