https://github.com/python/cpython/commit/563ab5cefe98d3a4a8f86eb46ca5aeaeaa5b120b
commit: 563ab5cefe98d3a4a8f86eb46ca5aeaeaa5b120b
branch: main
author: Barney Gale <[email protected]>
committer: barneygale <[email protected]>
date: 2025-03-16T06:11:20Z
summary:
GH-130614: pathlib ABCs: improve support for receiving path metadata (#131259)
In the private pathlib ABCs, replace `_WritablePath._write_info()` with
`_WritablePath._copy_from()`. This provides the target path object with
more control over the copying process, including support for querying and
setting metadata *before* the path is created.
Adjust `_ReadablePath.copy()` so that it forwards its keyword arguments to
`_WritablePath._copy_from()` of the target path object. This allows us to
remove the unimplemented *preserve_metadata* argument in the ABC method,
making it a `Path` exclusive.
files:
M Lib/pathlib/__init__.py
M Lib/pathlib/_os.py
M Lib/pathlib/types.py
diff --git a/Lib/pathlib/__init__.py b/Lib/pathlib/__init__.py
index 338016dbbbe972..c272ac4dddd40d 100644
--- a/Lib/pathlib/__init__.py
+++ b/Lib/pathlib/__init__.py
@@ -29,7 +29,7 @@
from pathlib._os import (
PathInfo, DirEntryInfo,
ensure_different_files, ensure_distinct_paths,
- copy_file, copy_info,
+ copyfile2, copyfileobj, magic_open, copy_info,
)
@@ -810,12 +810,6 @@ def write_text(self, data, encoding=None, errors=None,
newline=None):
with self.open(mode='w', encoding=encoding, errors=errors,
newline=newline) as f:
return f.write(data)
- def _write_info(self, info, follow_symlinks=True):
- """
- Write the given PathInfo to this path.
- """
- copy_info(info, self, follow_symlinks=follow_symlinks)
-
_remove_leading_dot = operator.itemgetter(slice(2, None))
_remove_trailing_slash = operator.itemgetter(slice(-1))
@@ -1100,18 +1094,21 @@ def replace(self, target):
target = self.with_segments(target)
return target
- def copy(self, target, follow_symlinks=True, preserve_metadata=False):
+ def copy(self, target, **kwargs):
"""
Recursively copy this file or directory tree to the given destination.
"""
if not hasattr(target, 'with_segments'):
target = self.with_segments(target)
ensure_distinct_paths(self, target)
- copy_file(self, target, follow_symlinks, preserve_metadata)
+ try:
+ copy_to_target = target._copy_from
+ except AttributeError:
+ raise TypeError(f"Target path is not writable: {target!r}") from
None
+ copy_to_target(self, **kwargs)
return target.joinpath() # Empty join to ensure fresh metadata.
- def copy_into(self, target_dir, *, follow_symlinks=True,
- preserve_metadata=False):
+ def copy_into(self, target_dir, **kwargs):
"""
Copy this file or directory tree into the given existing directory.
"""
@@ -1122,8 +1119,59 @@ def copy_into(self, target_dir, *, follow_symlinks=True,
target = target_dir / name
else:
target = self.with_segments(target_dir, name)
- return self.copy(target, follow_symlinks=follow_symlinks,
- preserve_metadata=preserve_metadata)
+ return self.copy(target, **kwargs)
+
+ def _copy_from(self, source, follow_symlinks=True,
preserve_metadata=False):
+ """
+ Recursively copy the given path to this path.
+ """
+ if not follow_symlinks and source.info.is_symlink():
+ self._copy_from_symlink(source, preserve_metadata)
+ elif source.info.is_dir():
+ children = source.iterdir()
+ os.mkdir(self)
+ for child in children:
+ self.joinpath(child.name)._copy_from(
+ child, follow_symlinks, preserve_metadata)
+ if preserve_metadata:
+ copy_info(source.info, self)
+ else:
+ self._copy_from_file(source, preserve_metadata)
+
+ def _copy_from_file(self, source, preserve_metadata=False):
+ ensure_different_files(source, self)
+ with magic_open(source, 'rb') as source_f:
+ with open(self, 'wb') as target_f:
+ copyfileobj(source_f, target_f)
+ if preserve_metadata:
+ copy_info(source.info, self)
+
+ if copyfile2:
+ # Use fast OS routine for local file copying where available.
+ _copy_from_file_fallback = _copy_from_file
+ def _copy_from_file(self, source, preserve_metadata=False):
+ try:
+ source = os.fspath(source)
+ except TypeError:
+ pass
+ else:
+ copyfile2(source, str(self))
+ return
+ self._copy_from_file_fallback(source, preserve_metadata)
+
+ if os.name == 'nt':
+ # If a directory-symlink is copied *before* its target, then
+ # os.symlink() incorrectly creates a file-symlink on Windows. Avoid
+ # this by passing *target_is_dir* to os.symlink() on Windows.
+ def _copy_from_symlink(self, source, preserve_metadata=False):
+ os.symlink(str(source.readlink()), self, source.info.is_dir())
+ if preserve_metadata:
+ copy_info(source.info, self, follow_symlinks=False)
+ else:
+ def _copy_from_symlink(self, source, preserve_metadata=False):
+ os.symlink(str(source.readlink()), self)
+ if preserve_metadata:
+ copy_info(source.info, self, follow_symlinks=False)
def move(self, target):
"""
diff --git a/Lib/pathlib/_os.py b/Lib/pathlib/_os.py
index 121b6d656a8f53..ee8657f427efbd 100644
--- a/Lib/pathlib/_os.py
+++ b/Lib/pathlib/_os.py
@@ -102,16 +102,16 @@ def _sendfile(source_fd, target_fd):
if _winapi and hasattr(_winapi, 'CopyFile2'):
- def _copyfile2(source, target):
+ def copyfile2(source, target):
"""
Copy from one file to another using CopyFile2 (Windows only).
"""
_winapi.CopyFile2(source, target, 0)
else:
- _copyfile2 = None
+ copyfile2 = None
-def _copyfileobj(source_f, target_f):
+def copyfileobj(source_f, target_f):
"""
Copy data from file-like object source_f to file-like object target_f.
"""
@@ -242,42 +242,6 @@ def ensure_different_files(source, target):
raise err
-def copy_file(source, target, follow_symlinks=True, preserve_metadata=False):
- """
- Recursively copy the given source ReadablePath to the given target
WritablePath.
- """
- info = source.info
- if not follow_symlinks and info.is_symlink():
- target.symlink_to(str(source.readlink()), info.is_dir())
- if preserve_metadata:
- target._write_info(info, follow_symlinks=False)
- elif info.is_dir():
- children = source.iterdir()
- target.mkdir()
- for src in children:
- dst = target.joinpath(src.name)
- copy_file(src, dst, follow_symlinks, preserve_metadata)
- if preserve_metadata:
- target._write_info(info)
- else:
- if _copyfile2:
- # Use fast OS routine for local file copying where available.
- try:
- source_p = os.fspath(source)
- target_p = os.fspath(target)
- except TypeError:
- pass
- else:
- _copyfile2(source_p, target_p)
- return
- ensure_different_files(source, target)
- with magic_open(source, 'rb') as source_f:
- with magic_open(target, 'wb') as target_f:
- _copyfileobj(source_f, target_f)
- if preserve_metadata:
- target._write_info(info)
-
-
def copy_info(info, target, follow_symlinks=True):
"""Copy metadata from the given PathInfo to the given local path."""
copy_times_ns = (
diff --git a/Lib/pathlib/types.py b/Lib/pathlib/types.py
index 9852bd4ff1e997..85dd9e5b2d6b9a 100644
--- a/Lib/pathlib/types.py
+++ b/Lib/pathlib/types.py
@@ -12,8 +12,8 @@
from abc import ABC, abstractmethod
from glob import _PathGlobber
+from pathlib._os import magic_open, ensure_distinct_paths,
ensure_different_files, copyfileobj
from pathlib import PurePath, Path
-from pathlib._os import magic_open, ensure_distinct_paths, copy_file
from typing import Optional, Protocol, runtime_checkable
@@ -332,18 +332,21 @@ def readlink(self):
"""
raise NotImplementedError
- def copy(self, target, follow_symlinks=True, preserve_metadata=False):
+ def copy(self, target, **kwargs):
"""
Recursively copy this file or directory tree to the given destination.
"""
if not hasattr(target, 'with_segments'):
target = self.with_segments(target)
ensure_distinct_paths(self, target)
- copy_file(self, target, follow_symlinks, preserve_metadata)
+ try:
+ copy_to_target = target._copy_from
+ except AttributeError:
+ raise TypeError(f"Target path is not writable: {target!r}") from
None
+ copy_to_target(self, **kwargs)
return target.joinpath() # Empty join to ensure fresh metadata.
- def copy_into(self, target_dir, *, follow_symlinks=True,
- preserve_metadata=False):
+ def copy_into(self, target_dir, **kwargs):
"""
Copy this file or directory tree into the given existing directory.
"""
@@ -354,8 +357,7 @@ def copy_into(self, target_dir, *, follow_symlinks=True,
target = target_dir / name
else:
target = self.with_segments(target_dir, name)
- return self.copy(target, follow_symlinks=follow_symlinks,
- preserve_metadata=preserve_metadata)
+ return self.copy(target, **kwargs)
class _WritablePath(_JoinablePath):
@@ -409,11 +411,25 @@ def write_text(self, data, encoding=None, errors=None,
newline=None):
with magic_open(self, mode='w', encoding=encoding, errors=errors,
newline=newline) as f:
return f.write(data)
- def _write_info(self, info, follow_symlinks=True):
- """
- Write the given PathInfo to this path.
- """
- pass
+ def _copy_from(self, source, follow_symlinks=True):
+ """
+ Recursively copy the given path to this path.
+ """
+ stack = [(source, self)]
+ while stack:
+ src, dst = stack.pop()
+ if not follow_symlinks and src.info.is_symlink():
+ dst.symlink_to(str(src.readlink()), src.info.is_dir())
+ elif src.info.is_dir():
+ children = src.iterdir()
+ dst.mkdir()
+ for child in children:
+ stack.append((child, dst.joinpath(child.name)))
+ else:
+ ensure_different_files(src, dst)
+ with magic_open(src, 'rb') as source_f:
+ with magic_open(dst, 'wb') as target_f:
+ copyfileobj(source_f, target_f)
_JoinablePath.register(PurePath)
_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3/lists/python-checkins.python.org/
Member address: [email protected]
