mdrs-client-python/mdrsclient/config.py

import configparser
import os
import threading
from typing import TYPE_CHECKING, Final, Protocol, runtime_checkable

import validators

from mdrsclient.exceptions import IllegalArgumentException
from mdrsclient.settings import CONFIG_DIRNAME
from mdrsclient.utils import FileLock


def normalize_url(url: str) -> str:
    """
    Check a remote URL and put it in the one form every client agrees on.

    `simple_host` is what lets a bare hostname through, so a development server on
    `localhost` is as acceptable as a deployment behind a domain name. The trailing
    slash goes because the URL is joined with a path that brings its own.
    """
    if not validators.url(url, simple_host=True, validate_scheme=lambda scheme: scheme in ("http", "https")):
        raise IllegalArgumentException("malformed URI sequence")
    return url.rstrip("/")


def build_download_url(base_url: str | None, path: str) -> str:
    """
    Join a remote base URL with a path the API returned.

    The API answers with a relative path and no leading separator, and a configuration
    written before the URL was normalised may still carry a trailing one.
    """
    if path.startswith(("http://", "https://")):
        return path
    return f"{(base_url or '').rstrip('/')}/{path.lstrip('/')}"


@runtime_checkable
class ConfigInterface(Protocol):
    remote: str

    if TYPE_CHECKING:
        # Declared for the type checker, because the class itself is passed around and
        # called: without this it describes something that takes no arguments to build.
        # Kept out of the running class, because a protocol that defines `__init__` loses
        # the guard that stops it being instantiated, and hands that empty `__init__` to
        # any implementation that does not write its own.
        def __init__(self, remote: str) -> None: ...

    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:
        url = normalize_url(url)
        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:
        url = normalize_url(url)
        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)