From f09eefc12a61590e72ee26325397aba041552f18 Mon Sep 17 00:00:00 2001 From: markbader Date: Thu, 12 Sep 2024 15:27:31 +0200 Subject: [PATCH] Implement metadata objects that fetch their data on access. --- webknossos/examples/accessing_metadata.py | 16 +-- webknossos/webknossos/dataset/_metadata.py | 104 ++++++++++++++++++ webknossos/webknossos/dataset/dataset.py | 21 ++-- .../webknossos/dataset/remote_folder.py | 15 +-- 4 files changed, 120 insertions(+), 36 deletions(-) create mode 100644 webknossos/webknossos/dataset/_metadata.py diff --git a/webknossos/examples/accessing_metadata.py b/webknossos/examples/accessing_metadata.py index 1b1e75002..c8a4f8fb1 100644 --- a/webknossos/examples/accessing_metadata.py +++ b/webknossos/examples/accessing_metadata.py @@ -2,25 +2,19 @@ def main() -> None: - with wk.webknossos_context( - url="https://webknossos.org/", - ): + with wk.webknossos_context(url="https://webknossos.org/"): l4_sample_dataset = wk.Dataset.open_remote("l4_sample") # Access the metadata of the dataset - dataset_metadata = l4_sample_dataset.metadata - print(dataset_metadata) + print(l4_sample_dataset.metadata) # Edit the metadata of the dataset - dataset_metadata["new_key"] = "new_value" - l4_sample_dataset.metadata = dataset_metadata + l4_sample_dataset.metadata["new_key"] = "new_value" # Access metadata of a folder - folder_metadata = l4_sample_dataset.folder.metadata - print(folder_metadata) + print(l4_sample_dataset.folder.metadata) # Edit the metadata of the folder - folder_metadata["new_folder_key"] = "new_folder_value" - l4_sample_dataset.folder.metadata = folder_metadata + l4_sample_dataset.folder.metadata["new_folder_key"] = "new_folder_value" if __name__ == "__main__": diff --git a/webknossos/webknossos/dataset/_metadata.py b/webknossos/webknossos/dataset/_metadata.py new file mode 100644 index 000000000..b649443c7 --- /dev/null +++ b/webknossos/webknossos/dataset/_metadata.py @@ -0,0 +1,104 @@ +from contextlib import contextmanager +from typing import ( + Any, + Dict, + Generator, + Iterator, + List, + Sequence, + TypeVar, + Union, +) + +from webknossos.client.api_client.models import ApiDataset, ApiFolder, ApiMetadata +from webknossos.utils import infer_metadata_type, parse_metadata_value + +_T = TypeVar("_T", bound="Metadata") + + +class Metadata(dict): + __slots__ = () + _api_path: str + _api_type: Any + + def __init__(self, _id: str, *args: Any, **kwargs: Dict[str, Any]) -> None: + if not self._api_path or not self._api_type: + raise NotImplementedError( + "This class is not meant to be used directly. Please use FolderMetadata or DatasetMetadata." + ) + super().__init__(*args, **kwargs) + self._id: str = _id + + @contextmanager + def _recent_metadata(self: _T) -> Generator[_T, None, None]: + from ..client.context import _get_api_client + + try: + client = _get_api_client() + full_object = client._get_json( + f"{self._api_path}{self._id}", + self._api_type, # type: ignore + ) + metadata: List[ApiMetadata] = full_object.metadata + if metadata is not None: + self = self.__class__( + self._id, + { + element.key: parse_metadata_value(element.value, element.type) + for element in metadata + }, + ) + else: + self = self.__class__(self._id) + yield self + finally: + api_metadata = [ + ApiMetadata(key=k, type=infer_metadata_type(v), value=v) + for k, v in self.items() + ] + + full_object.metadata = api_metadata + if self._api_type == ApiDataset: + client._patch_json(f"{self._api_path}{self._id}", full_object) + else: + client._put_json(f"{self._api_path}{self._id}", full_object) + + def __setitem__( + self, key: str, value: Union[str, int, float, Sequence[str]] + ) -> None: + with self._recent_metadata() as metadata: + super(Metadata, metadata).__setitem__(key, value) + + def __getitem__(self, key: str) -> Union[str, int, float, Sequence[str]]: + with self._recent_metadata() as metadata: + return super(Metadata, metadata).__getitem__(key) + + def __delitem__(self, key: str) -> None: + with self._recent_metadata() as metadata: + super(Metadata, metadata).__delitem__(key) + + def __contains__(self, key: object) -> bool: + with self._recent_metadata() as metadata: + return super(Metadata, metadata).__contains__(key) + + def __iter__(self) -> Iterator[Any]: + with self._recent_metadata() as metadata: + return super(Metadata, metadata).__iter__() + + def __len__(self) -> int: + with self._recent_metadata() as metadata: + return super(Metadata, metadata).__len__() + + def __repr__(self) -> str: + with self._recent_metadata() as metadata: + return f"{self.__class__.__name__}({super(Metadata, metadata).__repr__()})" + + +class FolderMetadata(Metadata): + _api_path = "/folders/" + _api_type = ApiFolder + + +class DatasetMetadata(Metadata): + _api_path = "/datasets/" + _api_type = ApiDataset diff --git a/webknossos/webknossos/dataset/dataset.py b/webknossos/webknossos/dataset/dataset.py index ea3c1679f..984376887 100644 --- a/webknossos/webknossos/dataset/dataset.py +++ b/webknossos/webknossos/dataset/dataset.py @@ -35,6 +35,7 @@ from numpy.typing import DTypeLike from upath import UPath +from webknossos.dataset._metadata import DatasetMetadata from webknossos.geometry.vec_int import VecIntLike from ..client.api_client.models import ApiDataset, ApiMetadata @@ -72,7 +73,6 @@ infer_metadata_type, is_fs_path, named_partial, - parse_metadata_value, rmtree, strip_trailing_slash, wait_and_ensure_success, @@ -2111,22 +2111,15 @@ def _update_dataset_info( ) @property - def metadata(self) -> Dict[str, Union[str, int, float, List[str]]]: - result = {} - if metadata := self._get_dataset_info().metadata: - for i in metadata: - value = parse_metadata_value(i.value, i.type) - if i.key in result: - warnings.warn( - f"The key {i.key} is a duplicate in the metadata. It is overwritten with last value." - ) - result[i.key] = value - - return result + def metadata(self) -> DatasetMetadata: + return DatasetMetadata(f"{self._organization_id}/{self._dataset_name}") @metadata.setter def metadata( - self, metadata: Optional[Dict[str, Union[str, int, float, Sequence[str]]]] + self, + metadata: Optional[ + Union[Dict[str, Union[str, int, float, Sequence[str]]], DatasetMetadata] + ], ) -> None: if metadata is not None: api_metadata = [ diff --git a/webknossos/webknossos/dataset/remote_folder.py b/webknossos/webknossos/dataset/remote_folder.py index f3efd567c..36bf389f7 100644 --- a/webknossos/webknossos/dataset/remote_folder.py +++ b/webknossos/webknossos/dataset/remote_folder.py @@ -2,7 +2,8 @@ import attr -from webknossos.utils import infer_metadata_type, parse_metadata_value +from webknossos.dataset._metadata import FolderMetadata +from webknossos.utils import infer_metadata_type from ..client.api_client.models import ApiFolder, ApiFolderWithParent, ApiMetadata @@ -53,16 +54,8 @@ def get_by_path(cls, path: str) -> "RemoteFolder": raise KeyError(f"Could not find folder {path}.") @property - def metadata(self) -> Dict[str, Union[str, int, float, List[str]]]: - from ..client.context import _get_api_client - - client = _get_api_client() - result = {} - if metadata := client._get_json(f"/folders/{self.id}", ApiFolder).metadata: - for i in metadata: - value = parse_metadata_value(i.value, i.type) - result[i.key] = value - return result + def metadata(self) -> FolderMetadata: + return FolderMetadata(self.id) @metadata.setter def metadata(