refactoring is on the way
diff --git a/wally/storage.py b/wally/storage.py
index 02de173..349995f 100644
--- a/wally/storage.py
+++ b/wally/storage.py
@@ -4,7 +4,7 @@
import os
import abc
-from typing import Any, Iterable, TypeVar, Type, IO, Tuple, Union, Dict, List
+from typing import Any, Iterable, TypeVar, Type, IO, Tuple, cast, List
class IStorable(metaclass=abc.ABCMeta):
@@ -46,11 +46,15 @@
pass
@abc.abstractmethod
+ def __delitem__(self, path: str) -> None:
+ pass
+
+ @abc.abstractmethod
def __contains__(self, path: str) -> bool:
pass
@abc.abstractmethod
- def list(self, path: str) -> Iterable[str]:
+ def list(self, path: str) -> Iterable[Tuple[bool, str]]:
pass
@abc.abstractmethod
@@ -78,35 +82,33 @@
if not os.path.isdir(self.root_path):
raise ValueError("No storage found at {!r}".format(root_path))
- def ensure_dir(self, path):
- os.makedirs(path, exist_ok=True)
-
- @abc.abstractmethod
def __setitem__(self, path: str, value: bytes) -> None:
path = os.path.join(self.root_path, path)
- self.ensure_dir(os.path.dirname(path))
+ os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "wb") as fd:
fd.write(value)
- @abc.abstractmethod
+ def __delitem__(self, path: str) -> None:
+ try:
+ os.unlink(path)
+ except FileNotFoundError:
+ pass
+
def __getitem__(self, path: str) -> bytes:
path = os.path.join(self.root_path, path)
with open(path, "rb") as fd:
return fd.read()
- @abc.abstractmethod
def __contains__(self, path: str) -> bool:
path = os.path.join(self.root_path, path)
return os.path.exists(path)
- @abc.abstractmethod
def list(self, path: str) -> Iterable[Tuple[bool, str]]:
path = os.path.join(self.root_path, path)
for entry in os.scandir(path):
if not entry.name in ('..', '.'):
yield entry.is_file(), entry.name
- @abc.abstractmethod
def get_stream(self, path: str, mode: str = "rb") -> IO:
path = os.path.join(self.root_path, path)
return open(path, mode)
@@ -114,41 +116,40 @@
class YAMLSerializer(ISerializer):
"""Serialize data to yaml"""
- pass
+ def pack(self, value: IStorable) -> bytes:
+ raise NotImplementedError()
-
-ISimpleStorable = Union[Dict, List, int, str, None, bool]
+ def unpack(self, data: bytes) -> IStorable:
+ raise NotImplementedError()
class Storage:
"""interface for storage"""
- def __init__(self, storage: ISimpleStorage, serializer: ISerializer):
+ def __init__(self, storage: ISimpleStorage, serializer: ISerializer) -> None:
self.storage = storage
self.serializer = serializer
def __setitem__(self, path: str, value: IStorable) -> None:
self.storage[path] = self.serializer.pack(value)
- @abc.abstractmethod
- def __getitem__(self, path: str) -> ISimpleStorable:
+ def __getitem__(self, path: str) -> IStorable:
return self.serializer.unpack(self.storage[path])
- @abc.abstractmethod
+ def __delitem__(self, path: str) -> None:
+ del self.storage[path]
+
def __contains__(self, path: str) -> bool:
return path in self.storage
- @abc.abstractmethod
def list(self, path: str) -> Iterable[Tuple[bool, str]]:
return self.storage.list(path)
- @abc.abstractmethod
- def load(self, path: str, obj_class: Type[ObjClass]) -> ObjClass:
- raw_val = self[path]
+ def construct(self, path: str, raw_val: IStorable, obj_class: Type[ObjClass]) -> ObjClass:
if obj_class in (int, str, dict, list, None):
if not isinstance(raw_val, obj_class):
raise ValueError("Can't load path {!r} into type {}. Real type is {}"
.format(path, obj_class, type(raw_val)))
- return raw_val
+ return cast(ObjClass, raw_val)
if not isinstance(raw_val, dict):
raise ValueError("Can't load path {!r} into python type. Raw value not dict".format(path))
@@ -157,14 +158,27 @@
raise ValueError("Can't load path {!r} into python type.".format(path) +
"Raw not all keys in raw value is strings")
- obj = ObjClass.__new__(ObjClass)
+ obj = obj_class.__new__(obj_class) # type: ObjClass
obj.__dict__.update(raw_val)
return obj
- @abc.abstractmethod
+ def load_list(self, path: str, obj_class: Type[ObjClass]) -> List[ObjClass]:
+ raw_val = self[path]
+ assert isinstance(raw_val, list)
+ return [self.construct(path, val, obj_class) for val in cast(list, raw_val)]
+
+ def load(self, path: str, obj_class: Type[ObjClass]) -> ObjClass:
+ return self.construct(path, self[path], obj_class)
+
def get_stream(self, path: str) -> IO:
return self.storage.get_stream(path)
+ def get(self, path: str, default: Any = None) -> Any:
+ try:
+ return self[path]
+ except KeyError:
+ return default
+
def make_storage(url: str, existing: bool = False) -> Storage:
return Storage(FSStorage(url, existing), YAMLSerializer())