dify/api/extensions/storage/base_storage.py
QuantumGhost a1c0bd7a1c
feat(api): Introduce workflow pause state management (#27298)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2025-10-30 14:41:09 +08:00

41 lines
1.2 KiB
Python

"""Abstract interface for file storage implementations."""
from abc import ABC, abstractmethod
from collections.abc import Generator
class BaseStorage(ABC):
"""Interface for file storage."""
@abstractmethod
def save(self, filename: str, data: bytes):
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
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")