common.storages.file_storage

FileStorage Objects

class FileStorage()

[view_source]

list_folder_files

def list_folder_files(relative_path: str, to_root: bool = True) -> List[str]

[view_source]

List all files in relative_path folder

Arguments:

• relative_path str - A path to folder, relative to storage root
• to_root bool, optional - If True returns paths to files in relation to root, if False, returns just file names. Defaults to True.

Returns:

• List[str] - A list of file names with optional path as per to_root parameter

link_hard_with_fallback

@staticmethod
def link_hard_with_fallback(external_file_path: str,
                            to_file_path: str) -> None

[view_source]

Try to create a hardlink and fallback to copying when filesystem doesn't support links

atomic_rename

def atomic_rename(from_relative_path: str, to_relative_path: str) -> None

[view_source]

Renames a path using os.rename which is atomic on POSIX, Windows and NFS v4.

Method falls back to non-atomic method in following cases:

1. On Windows when destination file exists
2. If underlying file system does not support atomic rename
3. All buckets mapped with FUSE are not atomic

rename_tree

def rename_tree(from_relative_path: str, to_relative_path: str) -> None

[view_source]

Renames a tree using os.rename if possible making it atomic

If we get 'too many open files': in that case `rename_tree_files is used

rename_tree_files

def rename_tree_files(from_relative_path: str, to_relative_path: str) -> None

[view_source]

Renames files in a tree recursively using os.rename.

atomic_import

def atomic_import(external_file_path: str,
                  to_folder: str,
                  new_file_name: Optional[str] = None) -> str

[view_source]

Moves a file at external_file_path into the to_folder effectively importing file into storage

Arguments:

• external_file_path - Path to file to be imported
• to_folder - Path to folder where file should be imported
• new_file_name - Optional new file name for the imported file, otherwise the original file name is used

Returns:

Path to imported file relative to storage root

open_zipsafe_ro

@staticmethod
def open_zipsafe_ro(path: str, mode: str = "r", **kwargs: Any) -> IO[Any]

[view_source]

Opens a file using gzip.open if it is a gzip file, otherwise uses open.

is_gzipped

@staticmethod
def is_gzipped(path: str) -> bool

[view_source]

Checks if file under path is gzipped by reading a header