aboutsummaryrefslogtreecommitdiff
path: root/src/hydrilla
diff options
context:
space:
mode:
Diffstat (limited to 'src/hydrilla')
-rw-r--r--src/hydrilla/item_infos.py173
-rw-r--r--src/hydrilla/server/malcontent.py4
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
)