mirror of
				https://github.com/langgenius/dify.git
				synced 2025-10-31 02:42:59 +00:00 
			
		
		
		
	
		
			
				
	
	
		
			41 lines
		
	
	
		
			1.1 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			41 lines
		
	
	
		
			1.1 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, 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
 | |
| 
 | |
|     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")
 | 
