Skip to content

File Storage Engine

Source code in reportconnectors/file_storage/engine/engine.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class StorageEngine(metaclass=ABCMeta):
    _b64_prefix = "b64_"
    _encoding = "utf-8"

    @abstractmethod
    def create_container(self, container_name: str, **kwargs) -> Optional[bool]:
        pass

    @abstractmethod
    def remove_container(self, container_name: str, include_files: bool = True, **kwargs) -> Optional[bool]:
        pass

    @abstractmethod
    def exists(self, container_name: str, file_name: Optional[str] = None) -> Optional[bool]:
        pass

    @abstractmethod
    def add_file(
        self, file: StorageFile, container_name: str, overwrite: bool = False, create_container: bool = True, **kwargs
    ) -> Optional[bool]:
        pass

    @abstractmethod
    def get_file(
        self,
        file_name: str,
        container_name: str,
        include_content: bool = True,
        include_metadata: bool = True,
        **kwargs,
    ) -> Optional[StorageFile]:
        pass

    @abstractmethod
    def get_file_link(self, file_name: str, container_name: str, **kwargs) -> Optional[str]:
        pass

    @abstractmethod
    def list_files(self, container_name: str, **kwargs) -> List[StorageFile]:
        pass

    @abstractmethod
    def delete_file(self, file_name: str, container_name: str, **kwargs) -> Optional[bool]:
        pass

    def __repr__(self):
        return f"{self.__class__.__name__}()"

    @staticmethod
    def _missing_text(container_name: str, file_name: Optional[str] = None):
        if file_name is None:
            return f"Container doesn't exists. ({container_name}/)"
        return f"File doesn't exists. ({container_name}/{file_name})"