| """ |
| Low-level OS functionality wrappers used by pathlib. |
| """ |
| |
| from errno import * |
| 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 _ficlone(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: |
| _ficlone = None |
| |
| |
| if posix and hasattr(posix, '_fcopyfile'): |
| def _fcopyfile(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) |
| else: |
| _fcopyfile = None |
| |
| |
| if hasattr(os, 'copy_file_range'): |
| def _copy_file_range(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 |
| else: |
| _copy_file_range = None |
| |
| |
| if hasattr(os, 'sendfile'): |
| def _sendfile(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: |
| _sendfile = None |
| |
| |
| if _winapi and hasattr(_winapi, 'CopyFile2'): |
| def copyfile(source, target): |
| """ |
| Copy from one file to another using CopyFile2 (Windows only). |
| """ |
| _winapi.CopyFile2(source, target, 0) |
| 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 _ficlone: |
| try: |
| _ficlone(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 _fcopyfile: |
| try: |
| _fcopyfile(source_fd, target_fd) |
| return |
| except OSError as err: |
| if err.errno not in (EINVAL, ENOTSUP): |
| raise err |
| if _copy_file_range: |
| try: |
| _copy_file_range(source_fd, target_fd) |
| return |
| except OSError as err: |
| if err.errno not in (ETXTBSY, EXDEV): |
| raise err |
| if _sendfile: |
| try: |
| _sendfile(source_fd, target_fd) |
| return |
| except OSError as err: |
| if err.errno != ENOTSOCK: |
| raise err |
| 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) |
| |
| |
| # 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 |
