Skip to content

Commit

Permalink
Implement metadata objects that fetch their data on access.
Browse files Browse the repository at this point in the history
  • Loading branch information
markbader committed Sep 12, 2024
1 parent cb5c1d6 commit f09eefc
Show file tree
Hide file tree
Showing 4 changed files with 120 additions and 36 deletions.
16 changes: 5 additions & 11 deletions webknossos/examples/accessing_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__":
Expand Down
104 changes: 104 additions & 0 deletions webknossos/webknossos/dataset/_metadata.py
Original file line number Diff line number Diff line change
@@ -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
21 changes: 7 additions & 14 deletions webknossos/webknossos/dataset/dataset.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -72,7 +73,6 @@
infer_metadata_type,
is_fs_path,
named_partial,
parse_metadata_value,
rmtree,
strip_trailing_slash,
wait_and_ensure_success,
Expand Down Expand Up @@ -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 = [
Expand Down
15 changes: 4 additions & 11 deletions webknossos/webknossos/dataset/remote_folder.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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(
Expand Down

0 comments on commit f09eefc

Please sign in to comment.