Source code for

"""Miscellaneous tools for operating on filesystems.

from __future__ import print_function
from __future__ import unicode_literals

import typing

from . import errors
from .errors import DirectoryNotEmpty
from .errors import ResourceNotFound
from .path import abspath
from .path import dirname
from .path import normpath
from .path import recursepath

if typing.TYPE_CHECKING:
    from typing import IO, List, Optional, Text
    from .base import FS

[docs]def remove_empty(fs, path): # type: (FS, Text) -> None """Remove all empty parents. Arguments: fs (FS): A filesystem instance. path (str): Path to a directory on the filesystem. """ path = abspath(normpath(path)) try: while path not in ("", "/"): fs.removedir(path) path = dirname(path) except DirectoryNotEmpty: pass
[docs]def copy_file_data(src_file, dst_file, chunk_size=None): # type: (IO, IO, Optional[int]) -> None """Copy data from one file object to another. Arguments: src_file (io.IOBase): File open for reading. dst_file (io.IOBase): File open for writing. chunk_size (int): Number of bytes to copy at a time (or `None` to use sensible default). """ _chunk_size = 1024 * 1024 if chunk_size is None else chunk_size read = write = dst_file.write # The 'or None' is so that it works with binary and text files for chunk in iter(lambda: read(_chunk_size) or None, None): write(chunk)
[docs]def get_intermediate_dirs(fs, dir_path): # type: (FS, Text) -> List[Text] """Get a list of non-existing intermediate directories. Arguments: fs (FS): A filesystem instance. dir_path (str): A path to a new directory on the filesystem. Returns: list: A list of non-existing paths. Raises: ~fs.errors.DirectoryExpected: If a path component references a file and not a directory. """ intermediates = [] with fs.lock(): for path in recursepath(abspath(dir_path), reverse=True): try: resource = fs.getinfo(path) except ResourceNotFound: intermediates.append(abspath(path)) else: if resource.is_dir: break raise errors.DirectoryExpected(dir_path) return intermediates[::-1][:-1]
[docs]def is_thread_safe(*filesystems): # type: (FS) -> bool """Check if all filesystems are thread-safe. Arguments: filesystems (FS): Filesystems instances to check. Returns: bool: if all filesystems are thread safe. """ return all(fs.getmeta().get("thread_safe", False) for fs in filesystems)