Use code from shutil.

python · barneygale · Jul 6, 2024 · Jun 20, 2024 · Jun 20, 2024 · Jun 21, 2024
commit f54925c8e086deb02c100f308e143678df12ded4
@@ -1562,11 +1562,6 @@ Other methods
    .. versionchanged:: 3.10
       The *follow_symlinks* parameter was added.
 
-   .. versionchanged:: 3.14
-      Raises :exc:`UnsupportedOperation` if *follow_symlinks* is false and
-      :func:`os.chmod` doesn't support this setting. In previous versions,
-      :exc:`NotImplementedError` was raised.
-
 .. method:: Path.expanduser()
 
    Return a new path with expanded ``~`` and ``~user`` constructs,
@@ -1603,10 +1598,6 @@ Other methods
    Like :meth:`Path.chmod` but, if the path points to a symbolic link, the
    symbolic link's mode is changed rather than its target's.
 
-   .. versionchanged:: 3.14
-      Raises :exc:`UnsupportedOperation` if :func:`os.chmod` doesn't support
-      setting *follow_symlinks* to false. In previous versions,
-      :exc:`NotImplementedError` was raised.
 
 .. method:: Path.owner(*, follow_symlinks=True)
 

@@ -15,7 +15,7 @@
 import operator
 import posixpath
 from glob import _GlobberBase, _no_recurse_symlinks
-from stat import S_ISDIR, S_ISLNK, S_ISREG, S_ISSOCK, S_ISBLK, S_ISCHR, S_ISFIFO, S_IMODE
+from stat import S_ISDIR, S_ISLNK, S_ISREG, S_ISSOCK, S_ISBLK, S_ISCHR, S_ISFIFO
 from ._os import copyfileobj
 
 
@@ -802,46 +802,28 @@ def copy(self, target, *, follow_symlinks=True, preserve_metadata=False):
             raise OSError(f"{self!r} and {target!r} are the same file")
         if not follow_symlinks and self.is_symlink():
             target.symlink_to(self.readlink())
-        else:
-            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
-        if preserve_metadata:
-            # Copy timestamps
-            st = self.stat(follow_symlinks=follow_symlinks)
-            try:
-                target._utime(ns=(st.st_atime_ns, st.st_mtime_ns),
-                              follow_symlinks=follow_symlinks)
-            except UnsupportedOperation:
-                pass
-            # Copy extended attributes (xattrs)
+            if preserve_metadata:
+                target._set_metadata(self._get_metadata(False), False)
+            return
+        with self.open('rb') as source_f:
             try:
-                for name in self._list_xattr(follow_symlinks=follow_symlinks):
-                    value = self._get_xattr(name, follow_symlinks=follow_symlinks)
-                    target._set_xattr(name, value, follow_symlinks=follow_symlinks)
-            except UnsupportedOperation:
-                pass
-            # Copy permissions (mode)
-            try:
-                target.chmod(mode=S_IMODE(st.st_mode),
-                             follow_symlinks=follow_symlinks)
-            except UnsupportedOperation:
-                pass
-            # Copy flags
-            if hasattr(st, 'st_flags'):
-                try:
-                    target._chflags(flags=st.st_flags,
-                                    follow_symlinks=follow_symlinks)
-                except (UnsupportedOperation, PermissionError):
-                    pass
+                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
+        if preserve_metadata:
+            target._set_metadata(self._get_metadata(True), True)
+
+    def _get_metadata(self, follow_symlinks):
+        return {}
+
+    def _set_metadata(self, metadata, follow_symlinks):
+        pass
 
     def rename(self, target):
         """
@@ -867,18 +849,6 @@ def replace(self, target):
         """
         raise UnsupportedOperation(self._unsupported_msg('replace()'))
 
-    def _utime(self, ns, *, follow_symlinks=True):
-        raise UnsupportedOperation(self._unsupported_msg('_utime()'))
-
-    def _list_xattr(self, *, follow_symlinks=True):
-        raise UnsupportedOperation(self._unsupported_msg('_list_xattr()'))
-
-    def _get_xattr(self, name, *, follow_symlinks=True):
-        raise UnsupportedOperation(self._unsupported_msg('_get_xattr()'))
-
-    def _set_xattr(self, name, value, *, follow_symlinks=True):
-        raise UnsupportedOperation(self._unsupported_msg('_set_xattr()'))
-
     def chmod(self, mode, *, follow_symlinks=True):
         """
         Change the permissions of the path, like os.chmod().
@@ -892,9 +862,6 @@ def lchmod(self, mode):
         """
         self.chmod(mode, follow_symlinks=False)
 
-    def _chflags(self, flags, *, follow_symlinks=True):
-        raise UnsupportedOperation(self._unsupported_msg('_chflags()'))
-
     def unlink(self, missing_ok=False):
         """
         Remove this file or link.

@@ -1,4 +1,3 @@
-import errno
 import io
 import ntpath
 import operator
@@ -19,7 +18,7 @@
     grp = None
 
 from ._abc import UnsupportedOperation, PurePathBase, PathBase
-from ._os import copyfile
+from ._os import copyfile, get_file_metadata, set_file_metadata
 
 
 __all__ = [
@@ -801,51 +800,14 @@ def copy(self, target, *, follow_symlinks=True, preserve_metadata=False):
                 raise
             copyfile(os.fspath(self), target, follow_symlinks)
 
-    def _utime(self, ns, *, follow_symlinks=True):
-        return os.utime(self, ns=ns, follow_symlinks=follow_symlinks)
-
-    if hasattr(os, 'listxattr'):
-        def _list_xattr(self, *, follow_symlinks=True):
-            try:
-                return os.listxattr(self, follow_symlinks=follow_symlinks)
-            except OSError as err:
-                if err.errno in (errno.ENOTSUP, errno.ENODATA, errno.EINVAL):
-                    raise UnsupportedOperation(str(err)) from None
-                raise
-
-        def _get_xattr(self, name, *, follow_symlinks=True):
-            try:
-                return os.getxattr(self, name, follow_symlinks=follow_symlinks)
-            except OSError as err:
-                if err.errno in (errno.ENOTSUP, errno.ENODATA, errno.EINVAL):
-                    raise UnsupportedOperation(str(err)) from None
-                raise
-
-        def _set_xattr(self, name, value, *, follow_symlinks=True):
-            try:
-                return os.setxattr(self, name, value, follow_symlinks=follow_symlinks)
-            except OSError as err:
-                if err.errno in (errno.ENOTSUP, errno.ENODATA, errno.EINVAL):
-                    raise UnsupportedOperation(str(err)) from None
-                raise
+    _get_metadata = get_file_metadata
+    _set_metadata = set_file_metadata
 
    def chmod(self, mode, *, follow_symlinks=True):
         """
         Change the permissions of the path, like os.chmod().
         """
-        try:
-            os.chmod(self, mode, follow_symlinks=follow_symlinks)
-        except NotImplementedError as err:
-            raise UnsupportedOperation(str(err)) from None
-
-    if hasattr(os, 'chflags'):
-        def _chflags(self, flags, *, follow_symlinks=True):
-            try:
-                os.chflags(self, flags, follow_symlinks=follow_symlinks)
-            except OSError as err:
-                if err.errno in (errno.ENOTSUP, errno.EOPNOTSUPP):
-                    raise UnsupportedOperation(str(err)) from None
-                raise
+        os.chmod(self, mode, follow_symlinks=follow_symlinks)
 
     def unlink(self, missing_ok=False):
         """

@@ -2,7 +2,7 @@
 Low-level OS functionality wrappers used by pathlib.
 """
 
-from errno import EBADF, EOPNOTSUPP, ETXTBSY, EXDEV
+from errno import *
 import os
 import stat
 import sys
@@ -157,3 +157,78 @@ def copyfileobj(source_f, target_f):
     write_target = target_f.write
     while buf := read_source(1024 * 1024):
         write_target(buf)
+
+
+def get_file_metadata(path, follow_symlinks):
+    if isinstance(path, os.DirEntry):
+        st = path.stat(follow_symlinks=follow_symlinks)
+    else:
+        st = os.stat(path, follow_symlinks=follow_symlinks)
+    result = {
+        'mode': stat.S_IMODE(st.st_mode),
+        'atime_ns': st.st_atime_ns,
+        'mtime_ns': st.st_mtime_ns,
+    }
+    if hasattr(os, 'listxattr'):
+        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
+    if hasattr(st, 'st_flags'):
+        result['flags'] = st.st_flags
+    return result
+
+
+def set_file_metadata(path, metadata, follow_symlinks):
+    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
+
+    lookup("utime")(path, ns=(metadata['atime_ns'], metadata['mtime_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:
+        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
+    try:
+        lookup("chmod")(path, metadata['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:
+        try:
+            lookup("chflags")(path, flags, follow_symlinks=follow_symlinks)
+        except OSError as why:
+            if why.errno not in (EOPNOTSUPP, ENOTSUP):
+                raise
diff --git a/Lib/shutil.py b/Lib/shutil.py
@@ -10,6 +10,7 @@
 import fnmatch
 import collections
 import errno
+from pathlib._os import get_file_metadata, set_file_metadata
 
 try:
     import zlib
@@ -317,34 +318,6 @@ def chmod_func(*args):
     st = stat_func(src)
     chmod_func(dst, stat.S_IMODE(st.st_mode))
 
-if hasattr(os, 'listxattr'):
-    def _copyxattr(src, dst, *, follow_symlinks=True):
-        """Copy extended filesystem attributes from `src` to `dst`.
-
-        Overwrite existing attributes.
-
-        If `follow_symlinks` is false, symlinks won't be followed.
-
-        """
-
-        try:
-            names = os.listxattr(src, follow_symlinks=follow_symlinks)
-        except OSError as e:
-            if e.errno not in (errno.ENOTSUP, errno.ENODATA, errno.EINVAL):
-                raise
-            return
-        for name in names:
-            try:
-                value = os.getxattr(src, name, follow_symlinks=follow_symlinks)
-                os.setxattr(dst, name, value, follow_symlinks=follow_symlinks)
-            except OSError as e:
-                if e.errno not in (errno.EPERM, errno.ENOTSUP, errno.ENODATA,
-                                   errno.EINVAL, errno.EACCES):
-                    raise
-else:
-    def _copyxattr(*args, **kwargs):
-        pass
-
 def copystat(src, dst, *, follow_symlinks=True):
     """Copy file metadata
 
@@ -359,57 +332,9 @@ def copystat(src, dst, *, follow_symlinks=True):
     """
     sys.audit("shutil.copystat", src, dst)
 
-    def _nop(*args, ns=None, follow_symlinks=None):
-        pass
-
     # follow symlinks (aka don't not follow symlinks)
     follow = follow_symlinks or not (_islink(src) and os.path.islink(dst))
-    if follow:
-        # 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
-
-    if isinstance(src, os.DirEntry):
-        st = src.stat(follow_symlinks=follow)
-    else:
-        st = lookup("stat")(src, follow_symlinks=follow)
-    mode = stat.S_IMODE(st.st_mode)
-    lookup("utime")(dst, ns=(st.st_atime_ns, st.st_mtime_ns),
-        follow_symlinks=follow)
-    # We must copy extended attributes before the file is (potentially)
-    # chmod()'ed read-only, otherwise setxattr() will error with -EACCES.
-    _copyxattr(src, dst, follow_symlinks=follow)
-    try:
-        lookup("chmod")(dst, mode, follow_symlinks=follow)
-    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
-    if hasattr(st, 'st_flags'):
-        try:
-            lookup("chflags")(dst, st.st_flags, follow_symlinks=follow)
-        except OSError as why:
-            for err in 'EOPNOTSUPP', 'ENOTSUP':
-                if hasattr(errno, err) and why.errno == getattr(errno, err):
-                    break
-            else:
-                raise
+    set_file_metadata(dst, get_file_metadata(src, follow), follow)
 
 def copy(src, dst, *, follow_symlinks=True):
     """Copy data and mode bits ("cp src dst"). Return the file's destination.