Spaces:
Running
Running
File size: 5,605 Bytes
21db53c |
1 2 3 4 5 6 7 8 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 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 |
import abc
import os
from typing import TypeVar, Generic, TypeAlias, Optional, AsyncGenerator
from app.Services.lifespan_service import LifespanService
FileMetaDataT = TypeVar('FileMetaDataT')
PathLikeType: TypeAlias = str | os.PathLike
LocalFilePathType: TypeAlias = PathLikeType | bytes
RemoteFilePathType: TypeAlias = PathLikeType
LocalFileMetaDataType: TypeAlias = FileMetaDataT
RemoteFileMetaDataType: TypeAlias = FileMetaDataT
class BaseStorage(LifespanService, abc.ABC, Generic[FileMetaDataT]):
def __init__(self):
self.static_dir: os.PathLike
self.thumbnails_dir: os.PathLike
self.deleted_dir: os.PathLike
self.file_metadata: FileMetaDataT
@abc.abstractmethod
async def is_exist(self,
remote_file: RemoteFilePathType) -> bool:
"""
Check if a remote_file exists.
:param remote_file: The file path relative to static_dir
:return: True if the file exists, False otherwise
"""
raise NotImplementedError
@abc.abstractmethod
async def size(self,
remote_file: RemoteFilePathType) -> int:
"""
Get the size of a file in static_dir
:param remote_file: The file path relative to static_dir
:return: file's size
"""
raise NotImplementedError
@abc.abstractmethod
async def url(self,
remote_file: RemoteFilePathType) -> str:
"""
Get the original URL of a file in static_dir.
This url will be placed in the payload field of the qdrant.
:param remote_file: The file path relative to static_dir
:return: file's "original URL"
"""
raise NotImplementedError
@abc.abstractmethod
async def presign_url(self,
remote_file: RemoteFilePathType,
expire_second: int = 3600) -> str:
"""
Get the presign URL of a file in static_dir.
:param remote_file: The file path relative to static_dir
:param expire_second: Valid time for presign url
:return: file's "presign URL"
"""
raise NotImplementedError
@abc.abstractmethod
async def fetch(self,
remote_file: RemoteFilePathType) -> bytes:
"""
Fetch a file from static_dir
:param remote_file: The file path relative to static_dir
:return: file's content
"""
raise NotImplementedError
@abc.abstractmethod
async def upload(self,
local_file: "LocalFilePathType",
remote_file: RemoteFilePathType) -> None:
"""
Move a local picture file to the static_dir.
:param local_file: The absolute path to the local file or bytes.
:param remote_file: The file path relative to static_dir
"""
raise NotImplementedError
@abc.abstractmethod
async def copy(self,
old_remote_file: RemoteFilePathType,
new_remote_file: RemoteFilePathType) -> None:
"""
Copy a file in static_dir.
:param old_remote_file: The file path relative to static_dir
:param new_remote_file: The file path relative to static_dir
"""
raise NotImplementedError
@abc.abstractmethod
async def move(self,
old_remote_file: RemoteFilePathType,
new_remote_file: RemoteFilePathType) -> None:
"""
Move a file in static_dir.
:param old_remote_file: The file path relative to static_dir
:param new_remote_file: The file path relative to static_dir
"""
raise NotImplementedError
@abc.abstractmethod
async def delete(self,
remote_file: RemoteFilePathType) -> None:
"""
Move a file in static_dir.
:param remote_file: The file path relative to static_dir
"""
raise NotImplementedError
@abc.abstractmethod
async def list_files(self,
path: RemoteFilePathType,
pattern: Optional[str] = "*",
batch_max_files: Optional[int] = None,
valid_extensions: Optional[set[str]] = None) \
-> AsyncGenerator[list[RemoteFilePathType], None]:
"""
Asynchronously generates a list of files from a given base directory path that match a specified pattern and set
of file extensions.
:param path: The relative base directory path from which relative to static_dir to start listing files.
:param pattern: A glob pattern to filter files based on their names. Defaults to '*' which selects all files.
:param batch_max_files: The maximum number of files to return. If None, all matching files are returned.
:param valid_extensions: An extra set of file extensions to include (e.g., {".jpg", ".png"}).
If None, files are not filtered by extension.
:return: An asynchronous generator yielding lists of RemoteFilePathType objects representing the matching files.
Usage example:
async for batch in list_files(base_path=".", pattern="*", max_files=100, valid_extensions={".jpg", ".png"}):
print(f"Batch: {batch}")
"""
raise NotImplementedError
@abc.abstractmethod
async def update_metadata(self,
local_file_metadata: LocalFileMetaDataType,
remote_file_metadata: RemoteFileMetaDataType) -> None:
raise NotImplementedError
|