Source code for fs.move

"""Functions for moving files between filesystems.
"""

from __future__ import print_function, unicode_literals

import typing

from ._pathcompat import commonpath
from .copy import copy_dir, copy_file
from .errors import FSError
from .opener import manage_fs
from .osfs import OSFS
from .path import frombase

if typing.TYPE_CHECKING:
    from typing import Text, Union

    from .base import FS


[docs]def move_fs( src_fs, # type: Union[Text, FS] dst_fs, # type:Union[Text, FS] workers=0, # type: int preserve_time=False, # type: bool ): # type: (...) -> None """Move the contents of a filesystem to another filesystem. Arguments: src_fs (FS or str): Source filesystem (instance or URL). dst_fs (FS or str): Destination filesystem (instance or URL). workers (int): Use `worker` threads to copy data, or ``0`` (default) for a single-threaded copy. preserve_time (bool): If `True`, try to preserve mtime of the resources (defaults to `False`). """ move_dir(src_fs, "/", dst_fs, "/", workers=workers, preserve_time=preserve_time)
[docs]def move_file( src_fs, # type: Union[Text, FS] src_path, # type: Text dst_fs, # type: Union[Text, FS] dst_path, # type: Text preserve_time=False, # type: bool cleanup_dst_on_error=True, # type: bool ): # type: (...) -> None """Move a file from one filesystem to another. Arguments: src_fs (FS or str): Source filesystem (instance or URL). src_path (str): Path to a file on ``src_fs``. dst_fs (FS or str): Destination filesystem (instance or URL). dst_path (str): Path to a file on ``dst_fs``. preserve_time (bool): If `True`, try to preserve mtime of the resources (defaults to `False`). cleanup_dst_on_error (bool): If `True`, tries to delete the file copied to ``dst_fs`` if deleting the file from ``src_fs`` fails (defaults to `True`). """ with manage_fs(src_fs, writeable=True) as _src_fs: with manage_fs(dst_fs, writeable=True, create=True) as _dst_fs: if _src_fs is _dst_fs: # Same filesystem, may be optimized _src_fs.move( src_path, dst_path, overwrite=True, preserve_time=preserve_time ) return if _src_fs.hassyspath(src_path) and _dst_fs.hassyspath(dst_path): # if both filesystems have a syspath we create a new OSFS from a # common parent folder and use it to move the file. try: src_syspath = _src_fs.getsyspath(src_path) dst_syspath = _dst_fs.getsyspath(dst_path) common = commonpath([src_syspath, dst_syspath]) if common: rel_src = frombase(common, src_syspath) rel_dst = frombase(common, dst_syspath) with _src_fs.lock(), _dst_fs.lock(): with OSFS(common) as base: base.move(rel_src, rel_dst, preserve_time=preserve_time) return # optimization worked, exit early except ValueError: # This is raised if we cannot find a common base folder. # In this case just fall through to the standard method. pass # Standard copy and delete with _src_fs.lock(), _dst_fs.lock(): copy_file( _src_fs, src_path, _dst_fs, dst_path, preserve_time=preserve_time, ) try: _src_fs.remove(src_path) except FSError as e: # if the source cannot be removed we delete the copy on the # destination if cleanup_dst_on_error: _dst_fs.remove(dst_path) raise e
[docs]def move_dir( src_fs, # type: Union[Text, FS] src_path, # type: Text dst_fs, # type: Union[Text, FS] dst_path, # type: Text workers=0, # type: int preserve_time=False, # type: bool ): # type: (...) -> None """Move a directory from one filesystem to another. Arguments: src_fs (FS or str): Source filesystem (instance or URL). src_path (str): Path to a directory on ``src_fs`` dst_fs (FS or str): Destination filesystem (instance or URL). dst_path (str): Path to a directory on ``dst_fs``. workers (int): Use ``worker`` threads to copy data, or ``0`` (default) for a single-threaded copy. preserve_time (bool): If `True`, try to preserve mtime of the resources (defaults to `False`). """ with manage_fs(src_fs, writeable=True) as _src_fs: with manage_fs(dst_fs, writeable=True, create=True) as _dst_fs: with _src_fs.lock(), _dst_fs.lock(): _dst_fs.makedir(dst_path, recreate=True) copy_dir( src_fs, src_path, dst_fs, dst_path, workers=workers, preserve_time=preserve_time, ) _src_fs.removetree(src_path)