Source code for xcube.core.zarrstore.diagnostic

# Copyright (c) 2018-2024 by xcube team and contributors
# Permissions are hereby granted under the terms of the MIT License:

from typing import Iterator, List


[docs] class DiagnosticZarrStore( """A diagnostic Zarr store used for testing and investigating behaviour of Zarr and xarray's Zarr backend. Args: store: Wrapped Zarr store. """ def __init__(self, store: = store self.records: List[str] = [] if hasattr(, "listdir"): self.listdir = self._listdir if hasattr(, "getsize"): self.getsize = self._getsize def _add_record(self, record: str): self.records.append(record) def _listdir(self, path: str = "") -> List[str]: self._add_record(f"listdir({path!r})") # noinspection PyUnresolvedReferences return def _getsize(self, key: str) -> int: self._add_record(f"getsize({key!r})") # noinspection PyUnresolvedReferences return
[docs] def keys(self): self._add_record("keys()") return
def __setitem__(self, key: str, value: bytes) -> None: self._add_record(f"__setitem__({key!r}, {type(value).__name__})"), value) def __delitem__(self, key: str) -> None: self._add_record(f"__delitem__({key!r})") def __getitem__(self, key: str) -> bytes: self._add_record(f"__getitem__({key!r})") return def __len__(self) -> int: self._add_record("__len__()") return def __iter__(self) -> Iterator[str]: self._add_record("__iter__()") return