GH-73991: Support preserving metadata in pathlib.Path.copy()

Add *preserve_metadata* keyword-only argument to `pathlib.Path.copy()`, defaulting to false. When set to true, we copy timestamps, permissions, extended attributes and flags where available, like `shutil.copystat()`. The argument has no effect on Windows, where metadata is always copied. In the pathlib ABCs we copy the file permissions with `PathBase.chmod()` where supported. In the future we might want to support a more generic public interface for copying metadata between different types of `PathBase` object, but it would be premature here.
python · barneygale · Jul 6, 2024 · Jun 20, 2024 · Jun 20, 2024 · Jun 21, 2024
commit ff025ff3e3563fd31c69cf9445d7726223437cce
@@ -1432,7 +1432,7 @@ Creating files and directories
 Copying, renaming and deleting
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
-.. method:: Path.copy(target, *, follow_symlinks=True)
+.. method:: Path.copy(target, *, follow_symlinks=True, preserve_metadata=False)
 
    Copy the contents of this file to the *target* file. If *target* specifies
    a file that already exists, it will be replaced.
@@ -1441,11 +1441,11 @@ Copying, renaming and deleting
    will be created as a symbolic link. If *follow_symlinks* is true and this
    file is a symbolic link, *target* will be a copy of the symlink target.
 
-   .. note::
-      This method uses operating system functionality to copy file content
-      efficiently. The OS might also copy some metadata, such as file
-      permissions. After the copy is complete, users may wish to call
-      :meth:`Path.chmod` to set the permissions of the target file.
+   If *preserve_metadata* is false (the default), only the file data is
+   guaranteed to be copied. Set *preserve_metadata* to true to ensure that the
+   file mode (permissions), flags, last access and modification times, and
+   extended attributes are all copied where supported. This argument has no
+   effect on Windows, where metadata is always preserved when copying.
 
    .. warning::
       On old builds of Windows (before Windows 10 build 19041), this method
@@ -1562,6 +1562,11 @@ 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,
@@ -1598,6 +1603,10 @@ 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
+from stat import S_ISDIR, S_ISLNK, S_ISREG, S_ISSOCK, S_ISBLK, S_ISCHR, S_ISFIFO, S_IMODE
 from ._os import copyfileobj
 
 
@@ -790,7 +790,7 @@ def mkdir(self, mode=0o777, parents=False, exist_ok=False):
         """
         raise UnsupportedOperation(self._unsupported_msg('mkdir()'))
 
-    def copy(self, target, follow_symlinks=True):
+    def copy(self, target, *, follow_symlinks=True, preserve_metadata=False):
         """
         Copy the contents of this file to the given target. If this file is a
         symlink and follow_symlinks is false, a symlink will be created at the
@@ -802,18 +802,46 @@ def copy(self, target, follow_symlinks=True):
             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())
-            return
-        with self.open('rb') as source_f:
+        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)
             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
+                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
 
     def rename(self, target):
         """
@@ -839,6 +867,18 @@ 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().
@@ -852,6 +892,9 @@ 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,3 +1,4 @@
+import errno
 import io
 import ntpath
 import operator
@@ -782,7 +783,7 @@ def mkdir(self, mode=0o777, parents=False, exist_ok=False):
                 raise
 
     if copyfile:
-        def copy(self, target, follow_symlinks=True):
+        def copy(self, target, *, follow_symlinks=True, preserve_metadata=False):
             """
             Copy the contents of this file to the given target. If this file is a
             symlink and follow_symlinks is false, a symlink will be created at the
@@ -794,15 +795,57 @@ def copy(self, target, follow_symlinks=True):
                 if isinstance(target, PathBase):
                     # Target is an instance of PathBase but not os.PathLike.
                     # Use generic implementation from PathBase.
-                    return PathBase.copy(self, target, follow_symlinks=follow_symlinks)
+                    return PathBase.copy(self, target,
+                                         follow_symlinks=follow_symlinks,
+                                         preserve_metadata=preserve_metadata)
                 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
+                return
+
+        def _get_xattr(self, name, *, follow_symlinks=True):
+            try:
+                return os.getxattr(self, name, follow_symlinks=follow_symlinks)
+            except OSError as err:
+                if e.errno in (errno.ENOTSUP, errno.ENODATA, errno.EINVAL):
+                    raise UnsupportedOperation(str(err)) from None
+                return
+
+        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 e.errno in (errno.ENOTSUP, errno.ENODATA, errno.EINVAL):
+                    raise UnsupportedOperation(str(err)) from None
+                return
+
     def chmod(self, mode, *, follow_symlinks=True):
         """
         Change the permissions of the path, like os.chmod().
         """
-        os.chmod(self, mode, follow_symlinks=follow_symlinks)
+        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
 
     def unlink(self, missing_ok=False):
         """

@@ -653,6 +653,46 @@ def test_open_unbuffered(self):
             self.assertIsInstance(f, io.RawIOBase)
             self.assertEqual(f.read().strip(), b"this is file A")
 
+    def test_copy_file_preserve_metadata(self):
+        base = self.cls(self.base)
+        source = base / 'fileA'
+        if hasattr(os, 'setxattr'):
+            os.setxattr(source, b'user.foo', b'42')
+        if hasattr(os, 'chmod'):
+            os.chmod(source, stat.S_IRWXU | stat.S_IRWXO)
+        if hasattr(os, 'chflags') and hasattr(stat, 'UF_NODUMP'):
+            os.chflags(source, stat.UF_NODUMP)
+        source_st = source.stat()
+        target = base / 'copyA'
+        source.copy(target, preserve_metadata=True)
+        self.assertTrue(target.exists())
+        self.assertEqual(source.read_text(), target.read_text())
+        target_st = target.stat()
+        if hasattr(os, 'getxattr'):
+            self.assertEqual(os.getxattr(target, b'user.foo'), b'42')
+        self.assertEqual(target_st.st_mode, source_st.st_mode)
+        if hasattr(source_st, 'st_flags'):
+            self.assertEqual(source_st.st_flags, target_st.st_flags)
+
+    @needs_symlinks
+    def test_copy_link_preserve_metadata(self):
+        base = self.cls(self.base)
+        source = base / 'linkA'
+        if hasattr(os, 'lchmod'):
+            os.lchmod(source, stat.S_IRWXU | stat.S_IRWXO)
+        if hasattr(os, 'lchflags') and hasattr(stat, 'UF_NODUMP'):
+            os.lchflags(source, stat.UF_NODUMP)
+        source_st = source.lstat()
+        target = base / 'copyA'
+        source.copy(target, follow_symlinks=False, preserve_metadata=True)
+        self.assertTrue(target.exists())
+        self.assertTrue(target.is_symlink())
+        self.assertEqual(source.readlink(), target.readlink())
+        target_st = target.lstat()
+        self.assertEqual(target_st.st_mode, source_st.st_mode)
+        if hasattr(source_st, 'st_flags'):
+            self.assertEqual(source_st.st_flags, target_st.st_flags)
+
     def test_resolve_nonexist_relative_issue38671(self):
         p = self.cls('non', 'exist')