diff options
Diffstat (limited to 'src/hydrilla')
| -rw-r--r-- | src/hydrilla/item_infos.py | 173 | ||||
| -rw-r--r-- | src/hydrilla/server/malcontent.py | 4 |
2 files changed, 147 insertions, 30 deletions
diff --git a/src/hydrilla/item_infos.py b/src/hydrilla/item_infos.py index 0e10871..525c6e3 100644 --- a/src/hydrilla/item_infos.py +++ b/src/hydrilla/item_infos.py @@ -43,7 +43,7 @@ import typing as t import dataclasses as dc from pathlib import Path, PurePosixPath -from abc import ABC +from abc import ABC, abstractmethod from immutables import Map @@ -419,6 +419,12 @@ def _load_item_info( ) +CategorizedInfoType = t.TypeVar( + 'CategorizedInfoType', + ResourceInfo, + MappingInfo +) + CategorizedType = t.TypeVar( 'CategorizedType', bound=Categorizable @@ -431,12 +437,16 @@ CategorizedUpdater = t.Callable[ CategoryKeyType = t.TypeVar('CategoryKeyType', bound=t.Hashable) -@dc.dataclass(frozen=True) -class CategorizedItemInfo(Categorizable, t.Generic[CategorizedType, CategoryKeyType]): +@dc.dataclass(frozen=True) # type: ignore[misc] +class CategorizedItemInfo( + ABC, + Categorizable, + t.Generic[CategorizedInfoType, CategorizedType, CategoryKeyType] +): """....""" SelfType = t.TypeVar( 'SelfType', - bound = 'CategorizedItemInfo[CategorizedType, CategoryKeyType]' + bound = 'CategorizedItemInfo[CategorizedInfoType, CategorizedType, CategoryKeyType]' ) uuid: t.Optional[str] = None @@ -481,72 +491,179 @@ class CategorizedItemInfo(Categorizable, t.Generic[CategorizedType, CategoryKeyT _initialized = self._initialized or updated is not None ) + @abstractmethod + def register(self: 'SelfType', info: CategorizedInfoType) -> 'SelfType': + ... + + @abstractmethod + def get_all(self: 'SelfType') -> t.Sequence[CategorizedInfoType]: + ... + def is_empty(self) -> bool: - """....""" return len(self.items) == 0 -VersionedType = t.TypeVar('VersionedType', ResourceInfo, MappingInfo) - class VersionedItemInfo( - CategorizedItemInfo[VersionedType, versions.VerTuple], - t.Generic[VersionedType] + CategorizedItemInfo[ + CategorizedInfoType, + CategorizedInfoType, + versions.VerTuple + ], + t.Generic[CategorizedInfoType] ): """Stores data of multiple versions of given resource/mapping.""" - SelfType = t.TypeVar('SelfType', bound='VersionedItemInfo[VersionedType]') + SelfType = t.TypeVar( + 'SelfType', + bound = 'VersionedItemInfo[CategorizedInfoType]' + ) - def register(self: 'SelfType', item_info: VersionedType) -> 'SelfType': + def register(self: 'SelfType', item_info: CategorizedInfoType) \ + -> 'SelfType': """ Make item info queryable by version. Perform sanity checks for uuid. """ return self._update(item_info.version, lambda old_info: item_info) - def unregister(self: 'SelfType', version: versions.VerTuple) -> 'SelfType': - """....""" - return self._update(version, lambda old_info: None) - @property def newest_version(self) -> versions.VerTuple: """....""" assert not self.is_empty() - return max(self.items.keys()) + return self.versions()[-1] @property - def newest_info(self) -> VersionedType: + def newest_info(self) -> CategorizedInfoType: """Find and return info of the newest version of item.""" return self.items[self.newest_version] - def get_by_ver(self, ver: t.Sequence[int]) -> t.Optional[VersionedType]: + def versions(self, reverse: bool = False) -> t.Sequence[versions.VerTuple]: + return sorted(self.items.keys(), reverse=reverse) + + def get_by_ver(self, ver: t.Sequence[int]) \ + -> t.Optional[CategorizedInfoType]: """ Find and return info of the specified version of the item (or None if absent). """ return self.items.get(versions.normalize(ver)) - def get_all(self) -> t.Iterable[VersionedType]: - """Generate item info for all its versions, from oldest to newest.""" - return [self.items[version] for version in sorted(self.items.keys())] + def get_all(self, reverse_versions: bool = False) \ + -> t.Sequence[CategorizedInfoType]: + """ + Generate item info for all its versions, from oldest to newest unless + the opposite is requested. + """ + versions = self.versions(reverse=reverse_versions) + return [self.items[ver] for ver in versions] VersionedResourceInfo = VersionedItemInfo[ResourceInfo] VersionedMappingInfo = VersionedItemInfo[MappingInfo] - VersionedItemInfoMap = Map[str, VersionedItemInfo] VersionedResourceInfoMap = Map[str, VersionedResourceInfo] VersionedMappingInfoMap = Map[str, VersionedMappingInfo] -def register_in_map( - map: Map[str, VersionedItemInfo[VersionedType]], - info: VersionedType -) -> Map[str, VersionedItemInfo[VersionedType]]: +def register_in_versioned_map( + map: Map[str, VersionedItemInfo[CategorizedInfoType]], + info: CategorizedInfoType +) -> Map[str, VersionedItemInfo[CategorizedInfoType]]: versioned_info = map.get(info.identifier, VersionedItemInfo()) return map.set(info.identifier, versioned_info.register(info)) + +class MultirepoItemInfo( + CategorizedItemInfo[ + CategorizedInfoType, + VersionedItemInfo[CategorizedInfoType], + tuple[str, int] + ], + t.Generic[CategorizedInfoType] +): + """ + Stores data of multiple versions of given resource/mapping that may come + from multiple repositories. + """ + SelfType = t.TypeVar( + 'SelfType', + bound = 'MultirepoItemInfo[CategorizedInfoType]' + ) + + def register(self: 'SelfType', item_info: CategorizedInfoType) \ + -> 'SelfType': + """ + Make item info queryable by repo and version. Perform sanity checks for + uuid. + """ + def update( + versioned: t.Optional[VersionedItemInfo[CategorizedInfoType]] + ) -> VersionedItemInfo[CategorizedInfoType]: + if versioned is None: + versioned = VersionedItemInfo() + return versioned.register(item_info) + + return self._update((item_info.repo, item_info.repo_iteration), update) + + @property + def default_info(self) -> CategorizedInfoType: + """ + Find and return info of one of the available options for the newest + version of item. + """ + assert not self.is_empty() + + return self.get_all(reverse_versions=True)[-1] + + def options(self, reverse: bool = False) -> t.Sequence[tuple[str, int]]: + return sorted( + self.items.keys(), + key = (lambda tuple: (tuple[0], 1 - tuple[1])), + reverse = reverse + ) + + def get_all( + self, + reverse_versions: bool = False, + reverse_repos: bool = False + ) -> t.Sequence[CategorizedInfoType]: + """ + Generate item info for all its versions and options, from oldest to + newest version and from. + """ + all_versions: set[versions.VerTuple] = set() + for versioned in self.items.values(): + all_versions.update(versioned.versions()) + + result = [] + + for version in sorted(all_versions, reverse=reverse_versions): + for option in self.options(reverse=reverse_repos): + info = self.items[option].get_by_ver(version) + if info is not None: + result.append(info) + + return result + +MultirepoResourceInfo = MultirepoItemInfo[ResourceInfo] +MultirepoMappingInfo = MultirepoItemInfo[MappingInfo] + + +MultirepoItemInfoMap = Map[str, MultirepoItemInfo] +MultirepoResourceInfoMap = Map[str, MultirepoResourceInfo] +MultirepoMappingInfoMap = Map[str, MultirepoMappingInfo] + +def register_in_multirepo_map( + map: Map[str, MultirepoItemInfo[CategorizedInfoType]], + info: CategorizedInfoType +) -> Map[str, MultirepoItemInfo[CategorizedInfoType]]: + multirepo_info = map.get(info.identifier, MultirepoItemInfo()) + + return map.set(info.identifier, multirepo_info.register(info)) + + def all_map_infos( - map: Map[str, VersionedItemInfo[VersionedType]] -) -> t.Iterator[VersionedType]: + map: Map[str, CategorizedItemInfo[CategorizedInfoType, t.Any, t.Any]] +) -> t.Iterator[CategorizedInfoType]: for versioned_info in map.values(): for item_info in versioned_info.get_all(): yield item_info diff --git a/src/hydrilla/server/malcontent.py b/src/hydrilla/server/malcontent.py index 9557161..5d51342 100644 --- a/src/hydrilla/server/malcontent.py +++ b/src/hydrilla/server/malcontent.py @@ -146,12 +146,12 @@ class Malcontent: self._check_package_files(item_info) if isinstance(item_info, item_infos.ResourceInfo): - self.resource_infos = item_infos.register_in_map( + self.resource_infos = item_infos.register_in_versioned_map( map = self.resource_infos, info = item_info ) else: - self.mapping_infos = item_infos.register_in_map( + self.mapping_infos = item_infos.register_in_versioned_map( map = self.mapping_infos, info = item_info ) |