# 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 ..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' @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: ... # 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 ) -> 'RepoRef': """....""" ... @abstractmethod def refresh(self) -> 'RepoIterationRef': """....""" ... @abstractmethod def get_display_info(self) -> 'RepoDisplayInfo': ... @dc.dataclass(frozen=True) class RepoDisplayInfo: ref: RepoRef name: str url: t.Optional[str] deleted: t.Optional[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.Iterable[RepoDisplayInfo]: ... @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class RepoIterationRef(Ref): """....""" pass @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class MappingRef(Ref): """....""" @abstractmethod def disable(self, state: 'HaketiloState') -> None: """....""" ... @abstractmethod def forget_enabled(self, state: 'HaketiloState') -> None: """....""" ... class MappingStore(Store[MappingRef]): pass @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) -> MappingDisplayInfo: ... @dc.dataclass(frozen=True) class MappingDisplayInfo: ref: MappingVersionRef info: item_infos.MappingInfo enabled: EnabledStatus is_orphan: bool class MappingVersionStore(Store[MappingVersionRef]): @abstractmethod def get_display_infos(self) -> t.Iterable[MappingDisplayInfo]: ... @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 add_repo(self, name: t.Optional[str], url: t.Optional[str]) \ -> RepoRef: """....""" ... @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: """....""" ...