return path.with_segments(str(path) + text)
+class CopyWorker:
+ """
+ Class that implements copying between path objects. An instance of this
+ class is available from the PathBase.copy property; it's made callable so
+ that PathBase.copy() can be treated as a method.
+
+ The target path's CopyWorker drives the process from its _create() method.
+ Files and directories are exchanged by calling methods on the source and
+ target paths, and metadata is exchanged by calling
+ source.copy._read_metadata() and target.copy._write_metadata().
+ """
+ __slots__ = ('_path',)
+
+ def __init__(self, path):
+ self._path = path
+
+ def __call__(self, target, follow_symlinks=True, dirs_exist_ok=False,
+ preserve_metadata=False):
+ """
+ Recursively copy this file or directory tree to the given destination.
+ """
+ if not isinstance(target, PathBase):
+ target = self._path.with_segments(target)
+
+ # Delegate to the target path's CopyWorker object.
+ return target.copy._create(self._path, follow_symlinks, dirs_exist_ok, preserve_metadata)
+
+ _readable_metakeys = frozenset()
+
+ def _read_metadata(self, metakeys, *, follow_symlinks=True):
+ """
+ Returns path metadata as a dict with string keys.
+ """
+ raise NotImplementedError
+
+ _writable_metakeys = frozenset()
+
+ def _write_metadata(self, metadata, *, follow_symlinks=True):
+ """
+ Sets path metadata from the given dict with string keys.
+ """
+ raise NotImplementedError
+
+ def _create(self, source, follow_symlinks, dirs_exist_ok, preserve_metadata):
+ self._ensure_distinct_path(source)
+ if preserve_metadata:
+ metakeys = self._writable_metakeys & source.copy._readable_metakeys
+ else:
+ metakeys = None
+ if not follow_symlinks and source.is_symlink():
+ self._create_symlink(source, metakeys)
+ elif source.is_dir():
+ self._create_dir(source, metakeys, follow_symlinks, dirs_exist_ok)
+ else:
+ self._create_file(source, metakeys)
+ return self._path
+
+ def _create_dir(self, source, metakeys, follow_symlinks, dirs_exist_ok):
+ """Copy the given directory to our path."""
+ children = list(source.iterdir())
+ self._path.mkdir(exist_ok=dirs_exist_ok)
+ for src in children:
+ dst = self._path.joinpath(src.name)
+ if not follow_symlinks and src.is_symlink():
+ dst.copy._create_symlink(src, metakeys)
+ elif src.is_dir():
+ dst.copy._create_dir(src, metakeys, follow_symlinks, dirs_exist_ok)
+ else:
+ dst.copy._create_file(src, metakeys)
+ if metakeys:
+ metadata = source.copy._read_metadata(metakeys)
+ if metadata:
+ self._write_metadata(metadata)
+
+ def _create_file(self, source, metakeys):
+ """Copy the given file to our path."""
+ self._ensure_different_file(source)
+ with source.open('rb') as source_f:
+ try:
+ with self._path.open('wb') as target_f:
+ copyfileobj(source_f, target_f)
+ except IsADirectoryError as e:
+ if not self._path.exists():
+ # Raise a less confusing exception.
+ raise FileNotFoundError(
+ f'Directory does not exist: {self._path}') from e
+ raise
+ if metakeys:
+ metadata = source.copy._read_metadata(metakeys)
+ if metadata:
+ self._write_metadata(metadata)
+
+ def _create_symlink(self, source, metakeys):
+ """Copy the given symbolic link to our path."""
+ self._path.symlink_to(source.readlink())
+ if metakeys:
+ metadata = source.copy._read_metadata(metakeys, follow_symlinks=False)
+ if metadata:
+ self._write_metadata(metadata, follow_symlinks=False)
+
+ def _ensure_different_file(self, source):
+ """
+ Raise OSError(EINVAL) if both paths refer to the same file.
+ """
+ pass
+
+ def _ensure_distinct_path(self, source):
+ """
+ Raise OSError(EINVAL) if the other path is within this path.
+ """
+ # Note: there is no straightforward, foolproof algorithm to determine
+ # if one directory is within another (a particularly perverse example
+ # would be a single network share mounted in one location via NFS, and
+ # in another location via CIFS), so we simply checks whether the
+ # other path is lexically equal to, or within, this path.
+ if source == self._path:
+ err = OSError(EINVAL, "Source and target are the same path")
+ elif source in self._path.parents:
+ err = OSError(EINVAL, "Source path is a parent of target path")
+ else:
+ return
+ err.filename = str(source)
+ err.filename2 = str(self._path)
+ raise err
+
+
class PurePathBase:
"""Base class for pure path objects.
except (OSError, ValueError):
return False
- def _ensure_different_file(self, other_path):
- """
- Raise OSError(EINVAL) if both paths refer to the same file.
- """
- pass
-
- def _ensure_distinct_path(self, other_path):
- """
- Raise OSError(EINVAL) if the other path is within this path.
- """
- # Note: there is no straightforward, foolproof algorithm to determine
- # if one directory is within another (a particularly perverse example
- # would be a single network share mounted in one location via NFS, and
- # in another location via CIFS), so we simply checks whether the
- # other path is lexically equal to, or within, this path.
- if self == other_path:
- err = OSError(EINVAL, "Source and target are the same path")
- elif self in other_path.parents:
- err = OSError(EINVAL, "Source path is a parent of target path")
- else:
- return
- err.filename = str(self)
- err.filename2 = str(other_path)
- raise err
-
def open(self, mode='r', buffering=-1, encoding=None,
errors=None, newline=None):
"""
"""
raise NotImplementedError
- def _symlink_to_target_of(self, link):
- """
- Make this path a symlink with the same target as the given link. This
- is used by copy().
- """
- self.symlink_to(link.readlink())
-
def mkdir(self, mode=0o777, parents=False, exist_ok=False):
"""
Create a new directory at this given path.
"""
raise NotImplementedError
- # Metadata keys supported by this path type.
- _readable_metadata = _writable_metadata = frozenset()
-
- def _read_metadata(self, keys=None, *, follow_symlinks=True):
- """
- Returns path metadata as a dict with string keys.
- """
- raise NotImplementedError
-
- def _write_metadata(self, metadata, *, follow_symlinks=True):
- """
- Sets path metadata from the given dict with string keys.
- """
- raise NotImplementedError
-
- def _copy_metadata(self, target, *, follow_symlinks=True):
- """
- Copies metadata (permissions, timestamps, etc) from this path to target.
- """
- # Metadata types supported by both source and target.
- keys = self._readable_metadata & target._writable_metadata
- if keys:
- metadata = self._read_metadata(keys, follow_symlinks=follow_symlinks)
- target._write_metadata(metadata, follow_symlinks=follow_symlinks)
-
- def _copy_file(self, target):
- """
- Copy the contents of this file to the given target.
- """
- self._ensure_different_file(target)
- with self.open('rb') as source_f:
- try:
- with target.open('wb') as target_f:
- copyfileobj(source_f, target_f)
- except IsADirectoryError as e:
- if not target.exists():
- # Raise a less confusing exception.
- raise FileNotFoundError(
- f'Directory does not exist: {target}') from e
- else:
- raise
-
- def copy(self, target, *, follow_symlinks=True, dirs_exist_ok=False,
- preserve_metadata=False):
- """
- Recursively copy this file or directory tree to the given destination.
- """
- if not isinstance(target, PathBase):
- target = self.with_segments(target)
- self._ensure_distinct_path(target)
- stack = [(self, target)]
- while stack:
- src, dst = stack.pop()
- if not follow_symlinks and src.is_symlink():
- dst._symlink_to_target_of(src)
- if preserve_metadata:
- src._copy_metadata(dst, follow_symlinks=False)
- elif src.is_dir():
- children = src.iterdir()
- dst.mkdir(exist_ok=dirs_exist_ok)
- stack.extend((child, dst.joinpath(child.name))
- for child in children)
- if preserve_metadata:
- src._copy_metadata(dst)
- else:
- src._copy_file(dst)
- if preserve_metadata:
- src._copy_metadata(dst)
- return target
+ copy = property(CopyWorker, doc=CopyWorker.__call__.__doc__)
def copy_into(self, target_dir, *, follow_symlinks=True,
dirs_exist_ok=False, preserve_metadata=False):
import os
import posixpath
import sys
-from errno import EINVAL, EXDEV
+from errno import *
from glob import _StringGlobber, _no_recurse_symlinks
from itertools import chain
-from stat import S_ISSOCK, S_ISBLK, S_ISCHR, S_ISFIFO
+from stat import S_IMODE, S_ISSOCK, S_ISBLK, S_ISCHR, S_ISFIFO
from _collections_abc import Sequence
try:
except ImportError:
grp = None
-from pathlib._os import (copyfile, file_metadata_keys, read_file_metadata,
- write_file_metadata)
-from pathlib._abc import PurePathBase, PathBase
+from pathlib._os import copyfile
+from pathlib._abc import CopyWorker, PurePathBase, PathBase
__all__ = [
return "<{}.parents>".format(type(self._path).__name__)
+class _LocalCopyWorker(CopyWorker):
+ """This object implements the Path.copy callable. Don't try to construct
+ it yourself."""
+ __slots__ = ()
+
+ _readable_metakeys = {'mode', 'times_ns'}
+ if hasattr(os.stat_result, 'st_flags'):
+ _readable_metakeys.add('flags')
+ if hasattr(os, 'listxattr'):
+ _readable_metakeys.add('xattrs')
+ _readable_metakeys = _writable_metakeys = frozenset(_readable_metakeys)
+
+ def _read_metadata(self, metakeys, *, follow_symlinks=True):
+ metadata = {}
+ if 'mode' in metakeys or 'times_ns' in metakeys or 'flags' in metakeys:
+ st = self._path.stat(follow_symlinks=follow_symlinks)
+ if 'mode' in metakeys:
+ metadata['mode'] = S_IMODE(st.st_mode)
+ if 'times_ns' in metakeys:
+ metadata['times_ns'] = st.st_atime_ns, st.st_mtime_ns
+ if 'flags' in metakeys:
+ metadata['flags'] = st.st_flags
+ if 'xattrs' in metakeys:
+ try:
+ metadata['xattrs'] = [
+ (attr, os.getxattr(self._path, attr, follow_symlinks=follow_symlinks))
+ for attr in os.listxattr(self._path, follow_symlinks=follow_symlinks)]
+ except OSError as err:
+ if err.errno not in (EPERM, ENOTSUP, ENODATA, EINVAL, EACCES):
+ raise
+ return metadata
+
+ def _write_metadata(self, metadata, *, follow_symlinks=True):
+ def _nop(*args, ns=None, follow_symlinks=None):
+ pass
+
+ if follow_symlinks:
+ # use the real function if it exists
+ def lookup(name):
+ return getattr(os, name, _nop)
+ else:
+ # use the real function only if it exists
+ # *and* it supports follow_symlinks
+ def lookup(name):
+ fn = getattr(os, name, _nop)
+ if fn in os.supports_follow_symlinks:
+ return fn
+ return _nop
+
+ times_ns = metadata.get('times_ns')
+ if times_ns is not None:
+ lookup("utime")(self._path, ns=times_ns, follow_symlinks=follow_symlinks)
+ # We must copy extended attributes before the file is (potentially)
+ # chmod()'ed read-only, otherwise setxattr() will error with -EACCES.
+ xattrs = metadata.get('xattrs')
+ if xattrs is not None:
+ for attr, value in xattrs:
+ try:
+ os.setxattr(self._path, attr, value, follow_symlinks=follow_symlinks)
+ except OSError as e:
+ if e.errno not in (EPERM, ENOTSUP, ENODATA, EINVAL, EACCES):
+ raise
+ mode = metadata.get('mode')
+ if mode is not None:
+ try:
+ lookup("chmod")(self._path, mode, follow_symlinks=follow_symlinks)
+ except NotImplementedError:
+ # if we got a NotImplementedError, it's because
+ # * follow_symlinks=False,
+ # * lchown() is unavailable, and
+ # * either
+ # * fchownat() is unavailable or
+ # * fchownat() doesn't implement AT_SYMLINK_NOFOLLOW.
+ # (it returned ENOSUP.)
+ # therefore we're out of options--we simply cannot chown the
+ # symlink. give up, suppress the error.
+ # (which is what shutil always did in this circumstance.)
+ pass
+ flags = metadata.get('flags')
+ if flags is not None:
+ try:
+ lookup("chflags")(self._path, flags, follow_symlinks=follow_symlinks)
+ except OSError as why:
+ if why.errno not in (EOPNOTSUPP, ENOTSUP):
+ raise
+
+ if copyfile:
+ # Use fast OS routine for local file copying where available.
+ def _create_file(self, source, metakeys):
+ """Copy the given file to the given target."""
+ try:
+ source = os.fspath(source)
+ except TypeError:
+ if not isinstance(source, PathBase):
+ raise
+ super()._create_file(source, metakeys)
+ else:
+ copyfile(source, os.fspath(self._path))
+
+ if os.name == 'nt':
+ # Windows: symlink target might not exist yet if we're copying several
+ # files, so ensure we pass is_dir to os.symlink().
+ def _create_symlink(self, source, metakeys):
+ """Copy the given symlink to the given target."""
+ self._path.symlink_to(source.readlink(), source.is_dir())
+ if metakeys:
+ metadata = source.copy._read_metadata(metakeys, follow_symlinks=False)
+ if metadata:
+ self._write_metadata(metadata, follow_symlinks=False)
+
+ def _ensure_different_file(self, source):
+ """
+ Raise OSError(EINVAL) if both paths refer to the same file.
+ """
+ try:
+ if not self._path.samefile(source):
+ return
+ except (OSError, ValueError):
+ return
+ err = OSError(EINVAL, "Source and target are the same file")
+ err.filename = str(source)
+ err.filename2 = str(self._path)
+ raise err
+
+
class PurePath(PurePathBase):
"""Base class for manipulating paths without I/O.
return (st.st_ino == other_st.st_ino and
st.st_dev == other_st.st_dev)
- def _ensure_different_file(self, other_path):
- """
- Raise OSError(EINVAL) if both paths refer to the same file.
- """
- try:
- if not self.samefile(other_path):
- return
- except (OSError, ValueError):
- return
- err = OSError(EINVAL, "Source and target are the same file")
- err.filename = str(self)
- err.filename2 = str(other_path)
- raise err
-
def open(self, mode='r', buffering=-1, encoding=None,
errors=None, newline=None):
"""
if not exist_ok or not self.is_dir():
raise
- _readable_metadata = _writable_metadata = file_metadata_keys
- _read_metadata = read_file_metadata
- _write_metadata = write_file_metadata
-
- if copyfile:
- def _copy_file(self, target):
- """
- Copy the contents of this file to the given target.
- """
- try:
- target = os.fspath(target)
- except TypeError:
- if not isinstance(target, PathBase):
- raise
- PathBase._copy_file(self, target)
- else:
- copyfile(os.fspath(self), target)
-
def chmod(self, mode, *, follow_symlinks=True):
"""
Change the permissions of the path, like os.chmod().
os.replace(self, target)
return self.with_segments(target)
+ copy = property(_LocalCopyWorker, doc=_LocalCopyWorker.__call__.__doc__)
+
def move(self, target):
"""
Recursively move this file or directory tree to the given destination.
"""
- self._ensure_different_file(target)
+ if not isinstance(target, PathBase):
+ target = self.with_segments(target)
+ target.copy._ensure_different_file(self)
try:
return self.replace(target)
- except TypeError:
- if not isinstance(target, PathBase):
- raise
except OSError as err:
if err.errno != EXDEV:
raise
f = f"{type(self).__name__}.symlink_to()"
raise UnsupportedOperation(f"{f} is unsupported on this system")
- if os.name == 'nt':
- def _symlink_to_target_of(self, link):
- """
- Make this path a symlink with the same target as the given link.
- This is used by copy().
- """
- self.symlink_to(link.readlink(), link.is_dir())
-
if hasattr(os, "link"):
def hardlink_to(self, target):
"""
from errno import *
import os
-import stat
import sys
try:
import fcntl
write_target = target_f.write
while buf := read_source(1024 * 1024):
write_target(buf)
-
-
-# Kinds of metadata supported by the operating system.
-file_metadata_keys = {'mode', 'times_ns'}
-if hasattr(os.stat_result, 'st_flags'):
- file_metadata_keys.add('flags')
-if hasattr(os, 'listxattr'):
- file_metadata_keys.add('xattrs')
-file_metadata_keys = frozenset(file_metadata_keys)
-
-
-def read_file_metadata(path, keys=None, *, follow_symlinks=True):
- """
- Returns local path metadata as a dict with string keys.
- """
- if keys is None:
- keys = file_metadata_keys
- assert keys.issubset(file_metadata_keys)
- result = {}
- for key in keys:
- if key == 'xattrs':
- try:
- result['xattrs'] = [
- (attr, os.getxattr(path, attr, follow_symlinks=follow_symlinks))
- for attr in os.listxattr(path, follow_symlinks=follow_symlinks)]
- except OSError as err:
- if err.errno not in (EPERM, ENOTSUP, ENODATA, EINVAL, EACCES):
- raise
- continue
- st = os.stat(path, follow_symlinks=follow_symlinks)
- if key == 'mode':
- result['mode'] = stat.S_IMODE(st.st_mode)
- elif key == 'times_ns':
- result['times_ns'] = st.st_atime_ns, st.st_mtime_ns
- elif key == 'flags':
- result['flags'] = st.st_flags
- return result
-
-
-def write_file_metadata(path, metadata, *, follow_symlinks=True):
- """
- Sets local path metadata from the given dict with string keys.
- """
- assert frozenset(metadata.keys()).issubset(file_metadata_keys)
-
- def _nop(*args, ns=None, follow_symlinks=None):
- pass
-
- if follow_symlinks:
- # use the real function if it exists
- def lookup(name):
- return getattr(os, name, _nop)
- else:
- # use the real function only if it exists
- # *and* it supports follow_symlinks
- def lookup(name):
- fn = getattr(os, name, _nop)
- if fn in os.supports_follow_symlinks:
- return fn
- return _nop
-
- times_ns = metadata.get('times_ns')
- if times_ns is not None:
- lookup("utime")(path, ns=times_ns, follow_symlinks=follow_symlinks)
- # We must copy extended attributes before the file is (potentially)
- # chmod()'ed read-only, otherwise setxattr() will error with -EACCES.
- xattrs = metadata.get('xattrs')
- if xattrs is not None:
- for attr, value in xattrs:
- try:
- os.setxattr(path, attr, value, follow_symlinks=follow_symlinks)
- except OSError as e:
- if e.errno not in (EPERM, ENOTSUP, ENODATA, EINVAL, EACCES):
- raise
- mode = metadata.get('mode')
- if mode is not None:
- try:
- lookup("chmod")(path, mode, follow_symlinks=follow_symlinks)
- except NotImplementedError:
- # if we got a NotImplementedError, it's because
- # * follow_symlinks=False,
- # * lchown() is unavailable, and
- # * either
- # * fchownat() is unavailable or
- # * fchownat() doesn't implement AT_SYMLINK_NOFOLLOW.
- # (it returned ENOSUP.)
- # therefore we're out of options--we simply cannot chown the
- # symlink. give up, suppress the error.
- # (which is what shutil always did in this circumstance.)
- pass
- flags = metadata.get('flags')
- if flags is not None:
- try:
- lookup("chflags")(path, flags, follow_symlinks=follow_symlinks)
- except OSError as why:
- if why.errno not in (EOPNOTSUPP, ENOTSUP):
- raise
projects / thirdparty / Python / cpython.git / commitdiff
