blob: e07d9b32dd921df08896c025f9f52c21b3e78dc6 [file] [log] [blame]
"""
This module contains interfaces for storage classes
"""
import os
import re
import abc
import shutil
import logging
from typing import Any, TypeVar, Type, IO, Tuple, cast, List, Dict, Iterable, Iterator
import yaml
try:
from yaml import CLoader as Loader, CDumper as Dumper # type: ignore
except ImportError:
from yaml import Loader, Dumper # type: ignore
from .common_types import IStorable
logger = logging.getLogger("wally")
class ISimpleStorage(metaclass=abc.ABCMeta):
"""interface for low-level storage, which doesn't support serialization
and can operate only on bytes"""
@abc.abstractmethod
def put(self, value: bytes, path: str) -> None:
pass
@abc.abstractmethod
def get(self, path: str) -> bytes:
pass
@abc.abstractmethod
def rm(self, path: str) -> None:
pass
@abc.abstractmethod
def sync(self) -> None:
pass
@abc.abstractmethod
def __contains__(self, path: str) -> bool:
pass
@abc.abstractmethod
def get_fd(self, path: str, mode: str = "rb+") -> IO:
pass
@abc.abstractmethod
def get_fname(self, path: str) -> str:
pass
@abc.abstractmethod
def sub_storage(self, path: str) -> 'ISimpleStorage':
pass
@abc.abstractmethod
def list(self, path: str) -> Iterator[Tuple[bool, str]]:
pass
class ITSStorage(metaclass=abc.ABCMeta):
"""interface for low-level storage, which doesn't support serialization
and can operate only on bytes"""
@abc.abstractmethod
def put(self, value: bytes, path: str) -> None:
pass
@abc.abstractmethod
def get(self, path: str) -> bytes:
pass
@abc.abstractmethod
def rm(self, path: str) -> None:
pass
@abc.abstractmethod
def sync(self) -> None:
pass
@abc.abstractmethod
def __contains__(self, path: str) -> bool:
pass
@abc.abstractmethod
def get_fd(self, path: str, mode: str = "rb+") -> IO:
pass
@abc.abstractmethod
def sub_storage(self, path: str) -> 'ISimpleStorage':
pass
@abc.abstractmethod
def list(self, path: str) -> Iterator[Tuple[bool, str]]:
pass
class ISerializer(metaclass=abc.ABCMeta):
"""Interface for serialization class"""
@abc.abstractmethod
def pack(self, value: IStorable) -> bytes:
pass
@abc.abstractmethod
def unpack(self, data: bytes) -> Any:
pass
class FSStorage(ISimpleStorage):
"""Store all data in files on FS"""
def __init__(self, root_path: str, existing: bool) -> None:
self.root_path = root_path
self.existing = existing
self.ignored = {'.', '..'}
def j(self, path: str) -> str:
return os.path.join(self.root_path, path)
def put(self, value: bytes, path: str) -> None:
jpath = self.j(path)
os.makedirs(os.path.dirname(jpath), exist_ok=True)
with open(jpath, "wb") as fd:
fd.write(value)
def get(self, path: str) -> bytes:
try:
with open(self.j(path), "rb") as fd:
return fd.read()
except FileNotFoundError as exc:
raise KeyError(path) from exc
def rm(self, path: str) -> None:
if os.path.isdir(path):
shutil.rmtree(path, ignore_errors=True)
elif os.path.exists(path):
os.unlink(path)
def __contains__(self, path: str) -> bool:
return os.path.exists(self.j(path))
def get_fname(self, path: str) -> str:
return self.j(path)
def get_fd(self, path: str, mode: str = "rb+") -> IO[bytes]:
jpath = self.j(path)
if "cb" == mode:
create_on_fail = True
mode = "rb+"
os.makedirs(os.path.dirname(jpath), exist_ok=True)
elif "ct" == mode:
create_on_fail = True
mode = "rt+"
os.makedirs(os.path.dirname(jpath), exist_ok=True)
else:
create_on_fail = False
try:
fd = open(jpath, mode)
except IOError:
if not create_on_fail:
raise
if 't' in mode:
fd = open(jpath, "wt")
else:
fd = open(jpath, "wb")
return cast(IO[bytes], fd)
def sub_storage(self, path: str) -> 'FSStorage':
return self.__class__(self.j(path), self.existing)
def sync(self):
pass
def list(self, path: str) -> Iterator[Tuple[bool, str]]:
path = self.j(path)
if not os.path.exists(path):
return
if not os.path.isdir(path):
raise OSError("{!r} is not a directory".format(path))
for fobj in os.scandir(path):
if fobj.path not in self.ignored:
if fobj.is_dir():
yield False, fobj.name
else:
yield True, fobj.name
class YAMLSerializer(ISerializer):
"""Serialize data to yaml"""
def pack(self, value: IStorable) -> bytes:
try:
return yaml.dump(value, Dumper=Dumper, encoding="utf8")
except Exception as exc:
raise ValueError("Can't pickle object {!r} to yaml".format(type(value))) from exc
def unpack(self, data: bytes) -> Any:
return yaml.load(data, Loader=Loader)
class SAFEYAMLSerializer(ISerializer):
"""Serialize data to yaml"""
def pack(self, value: IStorable) -> bytes:
try:
return yaml.safe_dump(value, encoding="utf8")
except Exception as exc:
raise ValueError("Can't pickle object {!r} to yaml".format(type(value))) from exc
def unpack(self, data: bytes) -> Any:
return yaml.safe_load(data)
ObjClass = TypeVar('ObjClass', bound=IStorable)
class _Raise:
pass
class Storage:
"""interface for storage"""
def __init__(self, sstorage: ISimpleStorage, serializer: ISerializer) -> None:
self.sstorage = sstorage
self.serializer = serializer
self.cache = {}
def sub_storage(self, *path: str) -> 'Storage':
fpath = "/".join(path)
return self.__class__(self.sstorage.sub_storage(fpath), self.serializer)
def put(self, value: Any, *path: str) -> None:
dct_value = cast(IStorable, value).raw() if isinstance(value, IStorable) else value
serialized = self.serializer.pack(dct_value) # type: ignore
fpath = "/".join(path)
self.sstorage.put(serialized, fpath)
def put_list(self, value: Iterable[IStorable], *path: str) -> None:
serialized = self.serializer.pack([obj.raw() for obj in value]) # type: ignore
fpath = "/".join(path)
self.sstorage.put(serialized, fpath)
def get(self, path: str, default: Any = _Raise) -> Any:
try:
vl = self.sstorage.get(path)
except:
if default is _Raise:
raise
return default
return self.serializer.unpack(vl)
def rm(self, *path: str) -> None:
fpath = "/".join(path)
self.sstorage.rm(fpath)
def __contains__(self, path: str) -> bool:
return path in self.sstorage
def put_raw(self, val: bytes, *path: str) -> str:
fpath = "/".join(path)
self.sstorage.put(val, fpath)
# TODO: dirty hack
return self.resolve_raw(fpath)
def resolve_raw(self, fpath) -> str:
return cast(FSStorage, self.sstorage).j(fpath)
def get_raw(self, *path: str) -> bytes:
return self.sstorage.get("/".join(path))
def append_raw(self, value: bytes, *path: str) -> None:
with self.sstorage.get_fd("/".join(path), "rb+") as fd:
fd.seek(0, os.SEEK_END)
fd.write(value)
def get_fd(self, path: str, mode: str = "r") -> IO:
return self.sstorage.get_fd(path, mode)
def get_fname(self, path: str) -> str:
return self.sstorage.get_fname(path)
def load_list(self, obj_class: Type[ObjClass], *path: str) -> List[ObjClass]:
path_s = "/".join(path)
if path_s not in self.cache:
raw_val = cast(List[Dict[str, Any]], self.get(path_s))
assert isinstance(raw_val, list)
self.cache[path_s] = [cast(ObjClass, obj_class.fromraw(val)) for val in raw_val]
return self.cache[path_s]
def load(self, obj_class: Type[ObjClass], *path: str) -> ObjClass:
path_s = "/".join(path)
if path_s not in self.cache:
self.cache[path_s] = cast(ObjClass, obj_class.fromraw(self.get(path_s)))
return self.cache[path_s]
def sync(self) -> None:
self.sstorage.sync()
def __enter__(self) -> 'Storage':
return self
def __exit__(self, x: Any, y: Any, z: Any) -> None:
self.sync()
def list(self, *path: str) -> Iterator[Tuple[bool, str]]:
return self.sstorage.list("/".join(path))
def _iter_paths(self,
root: str,
path_parts: List[str],
groups: Dict[str, str]) -> Iterator[Tuple[bool, str, Dict[str, str]]]:
curr = path_parts[0]
rest = path_parts[1:]
for is_file, name in self.list(root):
if rest and is_file:
continue
rr = re.match(pattern=curr + "$", string=name)
if rr:
if root:
path = root + "/" + name
else:
path = name
new_groups = rr.groupdict().copy()
new_groups.update(groups)
if rest:
yield from self._iter_paths(path, rest, new_groups)
else:
yield is_file, path, new_groups
def make_storage(url: str, existing: bool = False) -> Storage:
return Storage(FSStorage(url, existing), SAFEYAMLSerializer())