GH-73991: Add pathlib.Path.copy()

Doc/library/pathlib.rst

@@ -1477,6 +1477,24 @@ example because the path doesn't exist).
       available. In previous versions, :exc:`NotImplementedError` was raised.
 
 
+.. method:: Path.copy(target, *, follow_symlinks=True)
+
+   Copy the contents of this file to the *target* file. If *target* specifies
+   a file that already exists, it will be replaced.
+
+   If *follow_symlinks* is false, and this file is a symbolic link, *target*
+   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.
+
+   .. versionadded:: 3.14
+
+
 .. method:: Path.rename(target)
 
    Rename this file or directory to the given *target*, and return a new Path

Doc/whatsnew/3.14.rst

@@ -92,6 +92,12 @@ ast
 Added :func:`ast.compare` for comparing two ASTs.
 (Contributed by Batuhan Taskaya and Jeremy Hylton in :issue:`15987`.)
 
+pathlib
+-------
+
+* Add :meth:`pathlib.Path.copy`, which copies the content of one file to
+  another, like :func:`shutil.copyfile`.
+  (Contributed by Barney Gale in :gh:`73991`.)
 
 
 Optimizations

Lib/pathlib/_abc.py

@@ -15,6 +15,7 @@
 from glob import _Globber, _no_recurse_symlinks
 from errno import ENOTDIR, ELOOP
 from stat import S_ISDIR, S_ISLNK, S_ISREG, S_ISSOCK, S_ISBLK, S_ISCHR, S_ISFIFO
+from ._os import copyfileobj
 
 
 __all__ = ["UnsupportedOperation"]
@@ -535,6 +536,15 @@ def samefile(self, other_path):
         return (st.st_ino == other_st.st_ino and
                 st.st_dev == other_st.st_dev)
 
+    def _samefile_safe(self, other_path):
+        """
+        Like samefile(), but returns False rather than raising OSError.
+        """
+        try:
+            return self.samefile(other_path)
+        except (OSError, ValueError):
+            return False
+
     def open(self, mode='r', buffering=-1, encoding=None,
              errors=None, newline=None):
         """
@@ -783,6 +793,31 @@ def mkdir(self, mode=0o777, parents=False, exist_ok=False):
         """
         raise UnsupportedOperation(self._unsupported_msg('mkdir()'))
 
+    def copy(self, target, follow_symlinks=True):
+        """
+        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
+        target.
+        """
+        if not isinstance(target, PathBase):
+            target = self.with_segments(target)
+        if self._samefile_safe(target):
+            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:
+            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 rename(self, target):
         """
         Rename this path to the target path.

Lib/pathlib/_local.py

@@ -18,6 +18,7 @@
     grp = None
 
 from ._abc import UnsupportedOperation, PurePathBase, PathBase
+from ._os import copyfile
 
 
 __all__ = [
@@ -780,6 +781,23 @@ def mkdir(self, mode=0o777, parents=False, exist_ok=False):
             if not exist_ok or not self.is_dir():
                 raise
 
+    if copyfile:
+        def copy(self, target, follow_symlinks=True):
+            """
+            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
+            target.
+            """
+            try:
+                target = os.fspath(target)
+            except TypeError:
+                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)
+                raise
+            copyfile(os.fspath(self), target, follow_symlinks)
+
     def chmod(self, mode, *, follow_symlinks=True):
         """
         Change the permissions of the path, like os.chmod().

Lib/pathlib/_os.py

@@ -0,0 +1,158 @@
+"""
+Low-level OS functionality wrappers used by pathlib.
+"""
+
+from errno import EBADF, EOPNOTSUPP, ETXTBSY, EXDEV
+import os
+import stat
+import sys
+try:
+    import fcntl
+except ImportError:
+    fcntl = None
+try:
+    import posix
+except ImportError:
+    posix = None
+try:
+    import _winapi
+except ImportError:
+    _winapi = None
+
+
+def get_copy_blocksize(infd):
+    """Determine blocksize for fastcopying on Linux.
+    Hopefully the whole file will be copied in a single call.
+    The copying itself should be performed in a loop 'till EOF is
+    reached (0 return) so a blocksize smaller or bigger than the actual
+    file size should not make any difference, also in case the file
+    content changes while being copied.
+    """
+    try:
+        blocksize = max(os.fstat(infd).st_size, 2 ** 23)  # min 8 MiB
+    except OSError:
+        blocksize = 2 ** 27  # 128 MiB
+    # On 32-bit architectures truncate to 1 GiB to avoid OverflowError,
+    # see gh-82500.
+    if sys.maxsize < 2 ** 32:
+        blocksize = min(blocksize, 2 ** 30)
+    return blocksize
+
+
+if fcntl and hasattr(fcntl, 'FICLONE'):
+    def clonefd(source_fd, target_fd):
+        """
+        Perform a lightweight copy of two files, where the data blocks are
+        copied only when modified. This is known as Copy on Write (CoW),
+        instantaneous copy or reflink.
+        """
+        fcntl.ioctl(target_fd, fcntl.FICLONE, source_fd)
+else:
+    clonefd = None
+
+
+if posix and hasattr(posix, '_fcopyfile'):
+    def copyfd(source_fd, target_fd):
+        """
+        Copy a regular file content using high-performance fcopyfile(3)
+        syscall (macOS).
+        """
+        posix._fcopyfile(source_fd, target_fd, posix._COPYFILE_DATA)
+elif hasattr(os, 'copy_file_range'):
+    def copyfd(source_fd, target_fd):
+        """
+        Copy data from one regular mmap-like fd to another by using a
+        high-performance copy_file_range(2) syscall that gives filesystems
+        an opportunity to implement the use of reflinks or server-side
+        copy.
+        This should work on Linux >= 4.5 only.
+        """
+        blocksize = get_copy_blocksize(source_fd)
+        offset = 0
+        while True:
+            sent = os.copy_file_range(source_fd, target_fd, blocksize,
+                                      offset_dst=offset)
+            if sent == 0:
+                break  # EOF
+            offset += sent
+elif hasattr(os, 'sendfile'):
+    def copyfd(source_fd, target_fd):
+        """Copy data from one regular mmap-like fd to another by using
+        high-performance sendfile(2) syscall.
+        This should work on Linux >= 2.6.33 only.
+        """
+        blocksize = get_copy_blocksize(source_fd)
+        offset = 0
+        while True:
+            sent = os.sendfile(target_fd, source_fd, offset, blocksize)
+            if sent == 0:
+                break  # EOF
+            offset += sent
+else:
+    copyfd = None
+
+
+if _winapi and hasattr(_winapi, 'CopyFile2'):
+    def is_dirlink(path):
+        try:
+            st = os.lstat(path)
+        except OSError:
+            return False
+        return (st.st_file_attributes & stat.FILE_ATTRIBUTE_DIRECTORY and
+                st.st_reparse_tag == stat.IO_REPARSE_TAG_SYMLINK)
+
+    def copyfile(source, target, follow_symlinks):
+        """
+        Copy from one file to another using CopyFile2 (Windows only).
+        """
+        if follow_symlinks:
+            flags = 0
+        else:
+            flags = _winapi.COPY_FILE_COPY_SYMLINK
+            try:
+                _winapi.CopyFile2(source, target, flags)
+            except OSError as err:
+                # Check for ERROR_ACCESS_DENIED
+                if err.winerror != 5 or not is_dirlink(source):
+                    raise err
+            flags |= _winapi.COPY_FILE_DIRECTORY
+        _winapi.CopyFile2(source, target, flags)
+else:
+    copyfile = None
+
+
+def copyfileobj(source_f, target_f):
+    """
+    Copy data from file-like object source_f to file-like object target_f.
+    """
+    try:
+        source_fd = source_f.fileno()
+        target_fd = target_f.fileno()
+    except Exception:
+        pass  # Fall through to generic code.
+    else:
+        try:
+            # Use OS copy-on-write where available.
+            if clonefd:
+                try:
+                    clonefd(source_fd, target_fd)
+                    return
+                except OSError as err:
+                    if err.errno not in (EBADF, EOPNOTSUPP, ETXTBSY, EXDEV):
+                        raise err
+
+            # Use OS copy where available.
+            if copyfd:
+                copyfd(source_fd, target_fd)
+                return
+        except OSError as err:
+            # Produce more useful error messages.
+            err.filename = source_f.name
+            err.filename2 = target_f.name
+            raise err
+
+    # Last resort: copy with fileobj read() and write().
+    read_source = source_f.read
+    write_target = target_f.write
+    while buf := read_source(1024 * 1024):
+        write_target(buf)

Lib/test/test_pathlib/test_pathlib_abc.py

@@ -1696,6 +1696,88 @@ def test_write_text_with_newlines(self):
         self.assertEqual((p / 'fileA').read_bytes(),
                           b'abcde' + os_linesep_byte + b'fghlk' + os_linesep_byte + b'\rmnopq')
 
+    def test_copy_file(self):
+        base = self.cls(self.base)
+        source = base / 'fileA'
+        target = base / 'copyA'
+        source.copy(target)
+        self.assertTrue(target.exists())
+        self.assertEqual(source.read_text(), target.read_text())
+
+    def test_copy_directory(self):
+        base = self.cls(self.base)
+        source = base / 'dirA'
+        target = base / 'copyA'
+        with self.assertRaises(OSError):
+            source.copy(target)
+
+    @needs_symlinks
+    def test_copy_symlink_follow_symlinks_true(self):
+        base = self.cls(self.base)
+        source = base / 'linkA'
+        target = base / 'copyA'
+        source.copy(target)
+        self.assertTrue(target.exists())
+        self.assertFalse(target.is_symlink())
+        self.assertEqual(source.read_text(), target.read_text())
+
+    @needs_symlinks
+    def test_copy_symlink_follow_symlinks_false(self):
+        base = self.cls(self.base)
+        source = base / 'linkA'
+        target = base / 'copyA'
+        source.copy(target, follow_symlinks=False)
+        self.assertTrue(target.exists())
+        self.assertTrue(target.is_symlink())
+        self.assertEqual(source.readlink(), target.readlink())
+
+    @needs_symlinks
+    def test_copy_directory_symlink_follow_symlinks_false(self):
+        base = self.cls(self.base)
+        source = base / 'linkB'
+        target = base / 'copyA'
+        source.copy(target, follow_symlinks=False)
+        self.assertTrue(target.exists())
+        self.assertTrue(target.is_symlink())
+        self.assertEqual(source.readlink(), target.readlink())
+
+    def test_copy_to_existing_file(self):
+        base = self.cls(self.base)
+        source = base / 'fileA'
+        target = base / 'dirB' / 'fileB'
+        source.copy(target)
+        self.assertTrue(target.exists())
+        self.assertEqual(source.read_text(), target.read_text())
+
+    def test_copy_to_existing_directory(self):
+        base = self.cls(self.base)
+        source = base / 'fileA'
+        target = base / 'dirA'
+        with self.assertRaises(OSError):
+            source.copy(target)
+
+    @needs_symlinks
+    def test_copy_to_existing_symlink(self):
+        base = self.cls(self.base)
+        source = base / 'dirB' / 'fileB'
+        target = base / 'linkA'
+        real_target = base / 'fileA'
+        source.copy(target)
+        self.assertTrue(target.exists())
+        self.assertTrue(target.is_symlink())
+        self.assertTrue(real_target.exists())
+        self.assertFalse(real_target.is_symlink())
+        self.assertEqual(source.read_text(), real_target.read_text())
+
+    def test_copy_empty(self):
+        base = self.cls(self.base)
+        source = base / 'empty'
+        target = base / 'copyA'
+        source.write_bytes(b'')
+        source.copy(target)
+        self.assertTrue(target.exists())
+        self.assertEqual(target.read_bytes(), b'')
+
     def test_iterdir(self):
         P = self.cls
         p = P(self.base)

Misc/NEWS.d/next/Library/2024-05-15-01-36-08.gh-issue-73991.CGknDf.rst

@@ -0,0 +1,2 @@
+Add :meth:`pathlib.Path.copy`, which copies the content of one file to another,
+like :func:`shutil.copyfile`.

Modules/_winapi.c

@@ -3166,6 +3166,11 @@ static int winapi_exec(PyObject *m)
     #define COPY_FILE_REQUEST_COMPRESSED_TRAFFIC 0x10000000
 #endif
     WINAPI_CONSTANT(F_DWORD, COPY_FILE_REQUEST_COMPRESSED_TRAFFIC);
+#ifndef COPY_FILE_DIRECTORY
+    // Only defined in newer WinSDKs
+    #define COPY_FILE_DIRECTORY 0x00000080
+#endif
+    WINAPI_CONSTANT(F_DWORD, COPY_FILE_DIRECTORY);
 
     WINAPI_CONSTANT(F_DWORD, COPYFILE2_CALLBACK_CHUNK_STARTED);
     WINAPI_CONSTANT(F_DWORD, COPYFILE2_CALLBACK_CHUNK_FINISHED);