dlt.common.storages.file_storage
FileStorage Objects
class FileStorage()
touch_file
def touch_file(relative_path: str) -> None
Touch file, assumes single writer
list_folder_files
def list_folder_files(relative_path: str, to_root: bool = True) -> List[str]
List all files in relative_path folder
Arguments:
relative_pathstr - A path to folder, relative to storage rootto_rootbool, 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 perto_rootparameter
link_hard_with_fallback
@staticmethod
def link_hard_with_fallback(external_file_path: str,
to_file_path: str) -> None
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
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:
- On Windows when destination file exists
- If underlying file system does not support atomic rename
- All buckets mapped with FUSE are not atomic
rename_tree
def rename_tree(from_relative_path: str, to_relative_path: str) -> None
Renames a tree using os.rename if possible making it atomic
If we get 'too many open files' or a transient Windows access denied
during directory rename, rename_tree_files is used. On Windows, NTFS
may briefly hold directory handles after file operations inside the
tree, so EACCES/EPERM are retried with a short backoff before falling
back (same strategy as pip).
rename_tree_files
def rename_tree_files(from_relative_path: str, to_relative_path: str) -> None
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
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 importedto_folder- Path to folder where file should be importednew_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
is_path_in_storage
def is_path_in_storage(path: str) -> bool
Checks if a given path is below storage root, without checking for item existence
make_full_path_safe
def make_full_path_safe(path: str) -> str
Verifies that path is under storage root and then returns normalized absolute path
make_full_path
def make_full_path(path: str) -> str
Joins path with storage root. Intended for path known to be relative to storage root
open_zipsafe_ro
@staticmethod
def open_zipsafe_ro(path: str, mode: str = "r", **kwargs: Any) -> IO[Any]
Opens a file using gzip.open if it is a gzip file, otherwise uses open.
is_gzipped
@staticmethod
def is_gzipped(path: str) -> bool
Checks if file under path is gzipped by reading a header