# SPDX-License-Identifier: GPL-3.0-or-later # Haketilo proxy data and configuration (interface definition through abstract # class). # # This file is part of Hydrilla&Haketilo. # # Copyright (C) 2022 Wojtek Kosior # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . # # # I, Wojtek Kosior, thereby promise not to sue for violation of this # file's license. Although I request that you do not make use this code # in a proprietary program, I am not going to enforce this in court. """ This module defines API for keeping track of all settings, rules, mappings and resources. """ # Enable using with Python 3.7. from __future__ import annotations import dataclasses as dc import typing as t from pathlib import Path from abc import ABC, abstractmethod from enum import Enum from datetime import datetime from immutables import Map from ..exceptions import HaketiloException from ..versions import VerTuple from ..url_patterns import ParsedPattern from .. import item_infos class EnabledStatus(Enum): """ ENABLED - User wished to always apply given mapping when it matched. DISABLED - User wished to never apply given mapping. AUTO_ENABLED - User has not configured given mapping but it will still be used when automatic application of mappings is turned on. NO_MARK - User has not configured given mapping and it won't be used. """ ENABLED = 'E' DISABLED = 'D' AUTO_ENABLED = 'A' NO_MARK = 'N' class InstalledStatus(Enum): """ INSTALLED - Mapping's all files are present and mapping data is not going to be automatically removed. NOT_INSTALLED - Some of the mapping's files might be absent. Mapping can be automatically removed if it is orphaned. FAILED_TO_INSTALL - Same as "NOT_INSTALLED" but we additionally know that the last automatic attempt to install mapping's files from repository was unsuccessful. """ INSTALLED = 'I' NOT_INSTALLED = 'N' FAILED_TO_INSTALL = 'F' @dc.dataclass(frozen=True, unsafe_hash=True) class Ref: """....""" id: str def __post_init__(self): assert isinstance(self.id, str) RefType = t.TypeVar('RefType', bound=Ref) class Store(ABC, t.Generic[RefType]): @abstractmethod def get(self, id) -> RefType: ... class RepoNameInvalid(HaketiloException): pass class RepoNameTaken(HaketiloException): pass class RepoUrlInvalid(HaketiloException): pass class RepoCommunicationError(HaketiloException): pass @dc.dataclass class FileInstallationError(RepoCommunicationError): repo_id: str sha256: str class RepoApiVersionUnsupported(HaketiloException): pass # mypy needs to be corrected: # https://stackoverflow.com/questions/70999513/conflict-between-mix-ins-for-abstract-dataclasses/70999704#70999704 @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class RepoRef(Ref): """....""" @abstractmethod def remove(self) -> None: """....""" ... @abstractmethod def update( self, *, name: t.Optional[str] = None, url: t.Optional[str] = None ) -> None: """....""" ... @abstractmethod def refresh(self) -> 'RepoIterationRef': """....""" ... @abstractmethod def get_display_info(self) -> 'RepoDisplayInfo': ... @dc.dataclass(frozen=True) class RepoDisplayInfo: ref: RepoRef is_local_semirepo: bool name: str url: str deleted: bool last_refreshed: t.Optional[datetime] resource_count: int mapping_count: int class RepoStore(Store[RepoRef]): @abstractmethod def get_display_infos(self, include_deleted: bool = False) -> \ t.Sequence[RepoDisplayInfo]: ... @abstractmethod def add(self, name: str, url: str) -> RepoRef: ... @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class RepoIterationRef(Ref): """....""" pass @dc.dataclass(frozen=True) class MappingDisplayInfo: ref: 'MappingRef' identifier: str enabled: EnabledStatus active_version_ref: t.Optional['MappingVersionRef'] active_version_info: t.Optional[item_infos.MappingInfo] @dc.dataclass(frozen=True) class MappingVersionDisplayInfo: ref: 'MappingVersionRef' info: item_infos.MappingInfo installed: InstalledStatus is_active: bool is_orphan: bool is_local: bool mapping_enabled: EnabledStatus @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class MappingRef(Ref): """....""" @abstractmethod def get_version_display_infos(self) \ -> t.Sequence[MappingVersionDisplayInfo]: ... class MappingStore(Store[MappingRef]): @abstractmethod def get_display_infos(self) -> t.Sequence[MappingDisplayInfo]: ... @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class MappingVersionRef(Ref): """....""" @abstractmethod def update_status(self, new_status: EnabledStatus) -> None: """....""" ... @abstractmethod def get_display_info(self) -> MappingVersionDisplayInfo: ... class MappingVersionStore(Store[MappingVersionRef]): pass @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class ResourceRef(Ref): """....""" pass @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class ResourceVersionRef(Ref): """....""" pass @dc.dataclass(frozen=True) class PayloadKey: """....""" payload_ref: 'PayloadRef' mapping_identifier: str def __lt__(self, other: 'PayloadKey') -> bool: """....""" return self.mapping_identifier < other.mapping_identifier @dc.dataclass(frozen=True) class PayloadData: """....""" payload_ref: 'PayloadRef' explicitly_enabled: bool unique_token: str pattern_path_segments: tuple[str, ...] eval_allowed: bool cors_bypass_allowed: bool @dc.dataclass(frozen=True) class FileData: type: str name: str contents: bytes @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class PayloadRef(Ref): """....""" @abstractmethod def get_data(self) -> PayloadData: """....""" ... @abstractmethod def get_mapping(self) -> MappingVersionRef: """....""" ... @abstractmethod def get_script_paths(self) \ -> t.Iterable[t.Sequence[str]]: """....""" ... @abstractmethod def get_file_data(self, path: t.Sequence[str]) \ -> t.Optional[FileData]: """....""" ... class MappingUseMode(Enum): """ AUTO - Apply mappings except for those explicitly disabled. WHEN_ENABLED - Only apply mappings explicitly marked as enabled. Don't apply unmarked nor explicitly disabled mappings. QUESTION - Automatically apply mappings that are explicitly enabled. Ask whether to enable unmarked mappings. Don't apply explicitly disabled ones. """ AUTO = 'A' WHEN_ENABLED = 'W' QUESTION = 'Q' @dc.dataclass(frozen=True) class HaketiloGlobalSettings: """....""" mapping_use_mode: MappingUseMode default_allow_scripts: bool repo_refresh_seconds: int class MissingItemError(ValueError): """....""" pass class HaketiloState(ABC): """....""" @abstractmethod def import_packages(self, malcontent_path: Path) -> None: ... @abstractmethod def repo_store(self) -> RepoStore: """....""" ... @abstractmethod def get_repo_iteration(self, repo_iteration_id: str) -> RepoIterationRef: """....""" ... @abstractmethod def mapping_store(self) -> MappingStore: """....""" ... @abstractmethod def mapping_version_store(self) -> MappingVersionStore: """....""" ... @abstractmethod def get_resource(self, resource_id: str) -> ResourceRef: """....""" ... @abstractmethod def get_resource_version(self, resource_version_id: str) \ -> ResourceVersionRef: """....""" ... @abstractmethod def get_payload(self, payload_id: str) -> PayloadRef: """....""" ... @abstractmethod def get_settings(self) -> HaketiloGlobalSettings: """....""" ... @abstractmethod def update_settings( self, *, mapping_use_mode: t.Optional[MappingUseMode] = None, default_allow_scripts: t.Optional[bool] = None, repo_refresh_seconds: t.Optional[int] = None ) -> None: """....""" ...