2024-04-29 18:22:03 +08:00
|
|
|
"""Abstract interface for file storage implementations."""
|
2024-08-15 12:54:05 +08:00
|
|
|
|
2024-04-29 18:22:03 +08:00
|
|
|
from abc import ABC, abstractmethod
|
|
|
|
from collections.abc import Generator
|
|
|
|
|
|
|
|
|
|
|
|
class BaseStorage(ABC):
|
2024-08-15 12:54:05 +08:00
|
|
|
"""Interface for file storage."""
|
|
|
|
|
2024-04-29 18:22:03 +08:00
|
|
|
@abstractmethod
|
|
|
|
def save(self, filename, data):
|
|
|
|
raise NotImplementedError
|
|
|
|
|
|
|
|
@abstractmethod
|
|
|
|
def load_once(self, filename: str) -> bytes:
|
|
|
|
raise NotImplementedError
|
|
|
|
|
|
|
|
@abstractmethod
|
|
|
|
def load_stream(self, filename: str) -> Generator:
|
|
|
|
raise NotImplementedError
|
|
|
|
|
|
|
|
@abstractmethod
|
|
|
|
def download(self, filename, target_filepath):
|
|
|
|
raise NotImplementedError
|
|
|
|
|
|
|
|
@abstractmethod
|
|
|
|
def exists(self, filename):
|
|
|
|
raise NotImplementedError
|
|
|
|
|
|
|
|
@abstractmethod
|
|
|
|
def delete(self, filename):
|
|
|
|
raise NotImplementedError
|
2025-04-27 12:11:04 +09:00
|
|
|
|
|
|
|
def scan(self, path, files=True, directories=False) -> list[str]:
|
|
|
|
"""
|
|
|
|
Scan files and directories in the given path.
|
|
|
|
This method is implemented only in some storage backends.
|
|
|
|
If a storage backend doesn't support scanning, it will raise NotImplementedError.
|
|
|
|
"""
|
|
|
|
raise NotImplementedError("This storage backend doesn't support scanning")
|