forked from zarr-developers/zarr-python
/
core.py
85 lines (64 loc) · 2.64 KB
/
core.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
from __future__ import annotations
from pathlib import Path
from typing import Any, Optional, Tuple, Union
from zarr.v3.common import BytesLike
from zarr.v3.abc.store import Store
def _dereference_path(root: str, path: str) -> str:
assert isinstance(root, str)
assert isinstance(path, str)
root = root.rstrip("/")
path = f"{root}/{path}" if root != "" else path
path = path.rstrip("/")
return path
class StorePath:
store: Store
path: str
def __init__(self, store: Store, path: Optional[str] = None):
self.store = store
self.path = path or ""
@classmethod
def from_path(cls, pth: Path) -> StorePath:
# NOT SOLVED: This is instantiating an ABC + there is no from_path method
return cls(Store.from_path(pth)) # type: ignore
async def get(
self, byte_range: Optional[Tuple[int, Optional[int]]] = None
) -> Optional[BytesLike]:
return await self.store.get(self.path, byte_range)
async def set(self, value: BytesLike, byte_range: Optional[Tuple[int, int]] = None) -> None:
if byte_range is not None:
raise NotImplementedError("Store.set does not have partial writes yet")
await self.store.set(self.path, value)
async def delete(self) -> None:
await self.store.delete(self.path)
async def exists(self) -> bool:
return await self.store.exists(self.path)
def __truediv__(self, other: str) -> StorePath:
return self.__class__(self.store, _dereference_path(self.path, other))
def __str__(self) -> str:
return _dereference_path(str(self.store), self.path)
def __repr__(self) -> str:
return f"StorePath({self.store.__class__.__name__}, {repr(str(self))})"
def __eq__(self, other: Any) -> bool:
try:
if self.store == other.store and self.path == other.path:
return True
except Exception:
pass
return False
StoreLike = Union[Store, StorePath, Path, str]
def make_store_path(store_like: StoreLike) -> StorePath:
if isinstance(store_like, StorePath):
return store_like
elif isinstance(store_like, Store):
return StorePath(store_like)
# elif isinstance(store_like, Path):
# return StorePath(Store.from_path(store_like))
elif isinstance(store_like, str):
try:
from upath import UPath
# NOT SOLVED: Similar here, ABC instantiation + no from_path method
return StorePath(Store.from_path(UPath(store_like))) # type: ignore
except ImportError as e:
raise e
# return StorePath(LocalStore(Path(store_like)))
raise TypeError