class Storage(type):
get: t.Callable
flush: t.Callable
def __new__(cls, *_):
match storage := env.get('STORAGE', 'FILE_SYSTEM'):
case 'FILE_SYSTEM':
return super(Storage, cls).__new__(cls, FileSystemStorage)
case 'REDIS':
return super(Storage, cls).__new__(cls, RedisStorage)
case _:
raise StorageException(f'Unknown storage type: {storage}')
class FileSystemStorage(StorageInterface):
@exception_handler(OSError, StorageException)
def __init__(self) -> None:
self.root = DEFAULT_STORAGE_DIR
(Path.cwd() / self.root).mkdir(parents=True, exist_ok=True)
@exception_handler(OSError, StorageWriteException)
def put(self, name: str, data: bytes, path: str = '.') -> None:
filepath = Path(Path.cwd() / self.root / path / name)
filepath.parent.mkdir(parents=True, exist_ok=True)
with open(filepath, 'wb') as file:
file.write(data)
def put_many(self, files: dict[str, bytes], path: str = '.') -> None:
for name, data in files.items():
self.put(name, data, path)
@exception_handler(OSError, StorageReadException)
def get(self, name: str, path: str = '.') -> bytes:
filepath = Path(Path.cwd() / self.root / path / name)
if not self.exist(filepath):
raise StorageNotFoundException(f'File {name} not found in {filepath}')
with open(filepath, 'rb') as data:
return data.read()
def pop(self, name: str, path: str = '.') -> bytes:
data = self.get(name, path)
self.delete(name, path)
return data
@staticmethod
@exception_handler(OSError, StorageException)
def exist(path: Path) -> bool:
return path.exists()
@exception_handler(OSError, StorageException)
def delete(self, name: str, path: str = '.') -> None:
dir_path = Path(Path.cwd() / self.root / path)
os.remove(dir_path / name)
if not os.listdir(dir_path):
os.rmdir(dir_path)
@exception_handler(OSError, StorageException)
def flush(self, path: str = '.') -> None:
shutil.rmtree(Path(Path.cwd() / self.root / path), ignore_errors=True)
class StorageInterface:
def put(self, name: str, data: bytes, path: str = '.') -> None:
raise NotImplementedError()
def put_many(self, files: dict[str, bytes], path: str = '.') -> None:
raise NotImplementedError()
def get(self, name: str, path: str = '.') -> bytes:
raise NotImplementedError()
def pop(self, name: str, path: str = '.') -> bytes:
raise NotImplementedError()
@staticmethod
def exist(path: str) -> bool:
raise NotImplementedError()
def delete(self, name: str, path: str = '.') -> None:
raise NotImplementedError()
def flush(self, path: str = '.') -> None:
raise NotImplementedError()