codekingpro/portable-devtools

1"""2Low-level OS functionality wrappers used by pathlib.3"""4 5from errno import *6from io import TextIOWrapper, text_encoding7from stat import S_ISDIR, S_ISREG, S_ISLNK, S_IMODE8import os9import sys10try:11    import fcntl12except ImportError:13    fcntl = None14try:15    import posix16except ImportError:17    posix = None18try:19    import _winapi20except ImportError:21    _winapi = None22 23 24def _get_copy_blocksize(infd):25    """Determine blocksize for fastcopying on Linux.26    Hopefully the whole file will be copied in a single call.27    The copying itself should be performed in a loop 'till EOF is28    reached (0 return) so a blocksize smaller or bigger than the actual29    file size should not make any difference, also in case the file30    content changes while being copied.31    """32    try:33        blocksize = max(os.fstat(infd).st_size, 2 ** 23)  # min 8 MiB34    except OSError:35        blocksize = 2 ** 27  # 128 MiB36    # On 32-bit architectures truncate to 1 GiB to avoid OverflowError,37    # see gh-82500.38    if sys.maxsize < 2 ** 32:39        blocksize = min(blocksize, 2 ** 30)40    return blocksize41 42 43if fcntl and hasattr(fcntl, 'FICLONE'):44    def _ficlone(source_fd, target_fd):45        """46        Perform a lightweight copy of two files, where the data blocks are47        copied only when modified. This is known as Copy on Write (CoW),48        instantaneous copy or reflink.49        """50        fcntl.ioctl(target_fd, fcntl.FICLONE, source_fd)51else:52    _ficlone = None53 54 55if posix and hasattr(posix, '_fcopyfile'):56    def _fcopyfile(source_fd, target_fd):57        """58        Copy a regular file content using high-performance fcopyfile(3)59        syscall (macOS).60        """61        posix._fcopyfile(source_fd, target_fd, posix._COPYFILE_DATA)62else:63    _fcopyfile = None64 65 66if hasattr(os, 'copy_file_range'):67    def _copy_file_range(source_fd, target_fd):68        """69        Copy data from one regular mmap-like fd to another by using a70        high-performance copy_file_range(2) syscall that gives filesystems71        an opportunity to implement the use of reflinks or server-side72        copy.73        This should work on Linux >= 4.5 only.74        """75        blocksize = _get_copy_blocksize(source_fd)76        offset = 077        while True:78            sent = os.copy_file_range(source_fd, target_fd, blocksize,79                                      offset_dst=offset)80            if sent == 0:81                break  # EOF82            offset += sent83else:84    _copy_file_range = None85 86 87if hasattr(os, 'sendfile'):88    def _sendfile(source_fd, target_fd):89        """Copy data from one regular mmap-like fd to another by using90        high-performance sendfile(2) syscall.91        This should work on Linux >= 2.6.33 only.92        """93        blocksize = _get_copy_blocksize(source_fd)94        offset = 095        while True:96            sent = os.sendfile(target_fd, source_fd, offset, blocksize)97            if sent == 0:98                break  # EOF99            offset += sent100else:101    _sendfile = None102 103 104if _winapi and hasattr(_winapi, 'CopyFile2'):105    def copyfile2(source, target):106        """107        Copy from one file to another using CopyFile2 (Windows only).108        """109        _winapi.CopyFile2(source, target, 0)110else:111    copyfile2 = None112 113 114def copyfileobj(source_f, target_f):115    """116    Copy data from file-like object source_f to file-like object target_f.117    """118    try:119        source_fd = source_f.fileno()120        target_fd = target_f.fileno()121    except Exception:122        pass  # Fall through to generic code.123    else:124        try:125            # Use OS copy-on-write where available.126            if _ficlone:127                try:128                    _ficlone(source_fd, target_fd)129                    return130                except OSError as err:131                    if err.errno not in (EBADF, EOPNOTSUPP, ETXTBSY, EXDEV):132                        raise err133 134            # Use OS copy where available.135            if _fcopyfile:136                try:137                    _fcopyfile(source_fd, target_fd)138                    return139                except OSError as err:140                    if err.errno not in (EINVAL, ENOTSUP):141                        raise err142            if _copy_file_range:143                try:144                    _copy_file_range(source_fd, target_fd)145                    return146                except OSError as err:147                    if err.errno not in (ETXTBSY, EXDEV):148                        raise err149            if _sendfile:150                try:151                    _sendfile(source_fd, target_fd)152                    return153                except OSError as err:154                    if err.errno != ENOTSOCK:155                        raise err156        except OSError as err:157            # Produce more useful error messages.158            err.filename = source_f.name159            err.filename2 = target_f.name160            raise err161 162    # Last resort: copy with fileobj read() and write().163    read_source = source_f.read164    write_target = target_f.write165    while buf := read_source(1024 * 1024):166        write_target(buf)167 168 169def magic_open(path, mode='r', buffering=-1, encoding=None, errors=None,170               newline=None):171    """172    Open the file pointed to by this path and return a file object, as173    the built-in open() function does.174    """175    text = 'b' not in mode176    if text:177        # Call io.text_encoding() here to ensure any warning is raised at an178        # appropriate stack level.179        encoding = text_encoding(encoding)180    try:181        return open(path, mode, buffering, encoding, errors, newline)182    except TypeError:183        pass184    cls = type(path)185    mode = ''.join(sorted(c for c in mode if c not in 'bt'))186    if text:187        try:188            attr = getattr(cls, f'__open_{mode}__')189        except AttributeError:190            pass191        else:192            return attr(path, buffering, encoding, errors, newline)193    elif encoding is not None:194        raise ValueError("binary mode doesn't take an encoding argument")195    elif errors is not None:196        raise ValueError("binary mode doesn't take an errors argument")197    elif newline is not None:198        raise ValueError("binary mode doesn't take a newline argument")199 200    try:201        attr = getattr(cls, f'__open_{mode}b__')202    except AttributeError:203        pass204    else:205        stream = attr(path, buffering)206        if text:207            stream = TextIOWrapper(stream, encoding, errors, newline)208        return stream209 210    raise TypeError(f"{cls.__name__} can't be opened with mode {mode!r}")211 212 213def ensure_distinct_paths(source, target):214    """215    Raise OSError(EINVAL) if the other path is within this path.216    """217    # Note: there is no straightforward, foolproof algorithm to determine218    # if one directory is within another (a particularly perverse example219    # would be a single network share mounted in one location via NFS, and220    # in another location via CIFS), so we simply checks whether the221    # other path is lexically equal to, or within, this path.222    if source == target:223        err = OSError(EINVAL, "Source and target are the same path")224    elif source in target.parents:225        err = OSError(EINVAL, "Source path is a parent of target path")226    else:227        return228    err.filename = str(source)229    err.filename2 = str(target)230    raise err231 232 233def ensure_different_files(source, target):234    """235    Raise OSError(EINVAL) if both paths refer to the same file.236    """237    try:238        source_file_id = source.info._file_id239        target_file_id = target.info._file_id240    except AttributeError:241        if source != target:242            return243    else:244        try:245            if source_file_id() != target_file_id():246                return247        except (OSError, ValueError):248            return249    err = OSError(EINVAL, "Source and target are the same file")250    err.filename = str(source)251    err.filename2 = str(target)252    raise err253 254 255def copy_info(info, target, follow_symlinks=True):256    """Copy metadata from the given PathInfo to the given local path."""257    copy_times_ns = (258        hasattr(info, '_access_time_ns') and259        hasattr(info, '_mod_time_ns') and260        (follow_symlinks or os.utime in os.supports_follow_symlinks))261    if copy_times_ns:262        t0 = info._access_time_ns(follow_symlinks=follow_symlinks)263        t1 = info._mod_time_ns(follow_symlinks=follow_symlinks)264        os.utime(target, ns=(t0, t1), follow_symlinks=follow_symlinks)265 266    # We must copy extended attributes before the file is (potentially)267    # chmod()'ed read-only, otherwise setxattr() will error with -EACCES.268    copy_xattrs = (269        hasattr(info, '_xattrs') and270        hasattr(os, 'setxattr') and271        (follow_symlinks or os.setxattr in os.supports_follow_symlinks))272    if copy_xattrs:273        xattrs = info._xattrs(follow_symlinks=follow_symlinks)274        for attr, value in xattrs:275            try:276                os.setxattr(target, attr, value, follow_symlinks=follow_symlinks)277            except OSError as e:278                if e.errno not in (EPERM, ENOTSUP, ENODATA, EINVAL, EACCES):279                    raise280 281    copy_posix_permissions = (282        hasattr(info, '_posix_permissions') and283        (follow_symlinks or os.chmod in os.supports_follow_symlinks))284    if copy_posix_permissions:285        posix_permissions = info._posix_permissions(follow_symlinks=follow_symlinks)286        try:287            os.chmod(target, posix_permissions, follow_symlinks=follow_symlinks)288        except NotImplementedError:289            # if we got a NotImplementedError, it's because290            #   * follow_symlinks=False,291            #   * lchown() is unavailable, and292            #   * either293            #       * fchownat() is unavailable or294            #       * fchownat() doesn't implement AT_SYMLINK_NOFOLLOW.295            #         (it returned ENOSUP.)296            # therefore we're out of options--we simply cannot chown the297            # symlink.  give up, suppress the error.298            # (which is what shutil always did in this circumstance.)299            pass300 301    copy_bsd_flags = (302        hasattr(info, '_bsd_flags') and303        hasattr(os, 'chflags') and304        (follow_symlinks or os.chflags in os.supports_follow_symlinks))305    if copy_bsd_flags:306        bsd_flags = info._bsd_flags(follow_symlinks=follow_symlinks)307        try:308            os.chflags(target, bsd_flags, follow_symlinks=follow_symlinks)309        except OSError as why:310            if why.errno not in (EOPNOTSUPP, ENOTSUP):311                raise312 313 314class _PathInfoBase:315    __slots__ = ('_path', '_stat_result', '_lstat_result')316 317    def __init__(self, path):318        self._path = str(path)319 320    def __repr__(self):321        path_type = "WindowsPath" if os.name == "nt" else "PosixPath"322        return f"<{path_type}.info>"323 324    def _stat(self, *, follow_symlinks=True, ignore_errors=False):325        """Return the status as an os.stat_result, or None if stat() fails and326        ignore_errors is true."""327        if follow_symlinks:328            try:329                result = self._stat_result330            except AttributeError:331                pass332            else:333                if ignore_errors or result is not None:334                    return result335            try:336                self._stat_result = os.stat(self._path)337            except (OSError, ValueError):338                self._stat_result = None339                if not ignore_errors:340                    raise341            return self._stat_result342        else:343            try:344                result = self._lstat_result345            except AttributeError:346                pass347            else:348                if ignore_errors or result is not None:349                    return result350            try:351                self._lstat_result = os.lstat(self._path)352            except (OSError, ValueError):353                self._lstat_result = None354                if not ignore_errors:355                    raise356            return self._lstat_result357 358    def _posix_permissions(self, *, follow_symlinks=True):359        """Return the POSIX file permissions."""360        return S_IMODE(self._stat(follow_symlinks=follow_symlinks).st_mode)361 362    def _file_id(self, *, follow_symlinks=True):363        """Returns the identifier of the file."""364        st = self._stat(follow_symlinks=follow_symlinks)365        return st.st_dev, st.st_ino366 367    def _access_time_ns(self, *, follow_symlinks=True):368        """Return the access time in nanoseconds."""369        return self._stat(follow_symlinks=follow_symlinks).st_atime_ns370 371    def _mod_time_ns(self, *, follow_symlinks=True):372        """Return the modify time in nanoseconds."""373        return self._stat(follow_symlinks=follow_symlinks).st_mtime_ns374 375    if hasattr(os.stat_result, 'st_flags'):376        def _bsd_flags(self, *, follow_symlinks=True):377            """Return the flags."""378            return self._stat(follow_symlinks=follow_symlinks).st_flags379 380    if hasattr(os, 'listxattr'):381        def _xattrs(self, *, follow_symlinks=True):382            """Return the xattrs as a list of (attr, value) pairs, or an empty383            list if extended attributes aren't supported."""384            try:385                return [386                    (attr, os.getxattr(self._path, attr, follow_symlinks=follow_symlinks))387                    for attr in os.listxattr(self._path, follow_symlinks=follow_symlinks)]388            except OSError as err:389                if err.errno not in (EPERM, ENOTSUP, ENODATA, EINVAL, EACCES):390                    raise391                return []392 393 394class _WindowsPathInfo(_PathInfoBase):395    """Implementation of pathlib.types.PathInfo that provides status396    information for Windows paths. Don't try to construct it yourself."""397    __slots__ = ('_exists', '_is_dir', '_is_file', '_is_symlink')398 399    def exists(self, *, follow_symlinks=True):400        """Whether this path exists."""401        if not follow_symlinks and self.is_symlink():402            return True403        try:404            return self._exists405        except AttributeError:406            if os.path.exists(self._path):407                self._exists = True408                return True409            else:410                self._exists = self._is_dir = self._is_file = False411                return False412 413    def is_dir(self, *, follow_symlinks=True):414        """Whether this path is a directory."""415        if not follow_symlinks and self.is_symlink():416            return False417        try:418            return self._is_dir419        except AttributeError:420            if os.path.isdir(self._path):421                self._is_dir = self._exists = True422                return True423            else:424                self._is_dir = False425                return False426 427    def is_file(self, *, follow_symlinks=True):428        """Whether this path is a regular file."""429        if not follow_symlinks and self.is_symlink():430            return False431        try:432            return self._is_file433        except AttributeError:434            if os.path.isfile(self._path):435                self._is_file = self._exists = True436                return True437            else:438                self._is_file = False439                return False440 441    def is_symlink(self):442        """Whether this path is a symbolic link."""443        try:444            return self._is_symlink445        except AttributeError:446            self._is_symlink = os.path.islink(self._path)447            return self._is_symlink448 449 450class _PosixPathInfo(_PathInfoBase):451    """Implementation of pathlib.types.PathInfo that provides status452    information for POSIX paths. Don't try to construct it yourself."""453    __slots__ = ()454 455    def exists(self, *, follow_symlinks=True):456        """Whether this path exists."""457        st = self._stat(follow_symlinks=follow_symlinks, ignore_errors=True)458        if st is None:459            return False460        return True461 462    def is_dir(self, *, follow_symlinks=True):463        """Whether this path is a directory."""464        st = self._stat(follow_symlinks=follow_symlinks, ignore_errors=True)465        if st is None:466            return False467        return S_ISDIR(st.st_mode)468 469    def is_file(self, *, follow_symlinks=True):470        """Whether this path is a regular file."""471        st = self._stat(follow_symlinks=follow_symlinks, ignore_errors=True)472        if st is None:473            return False474        return S_ISREG(st.st_mode)475 476    def is_symlink(self):477        """Whether this path is a symbolic link."""478        st = self._stat(follow_symlinks=False, ignore_errors=True)479        if st is None:480            return False481        return S_ISLNK(st.st_mode)482 483 484PathInfo = _WindowsPathInfo if os.name == 'nt' else _PosixPathInfo485 486 487class DirEntryInfo(_PathInfoBase):488    """Implementation of pathlib.types.PathInfo that provides status489    information by querying a wrapped os.DirEntry object. Don't try to490    construct it yourself."""491    __slots__ = ('_entry',)492 493    def __init__(self, entry):494        super().__init__(entry.path)495        self._entry = entry496 497    def _stat(self, *, follow_symlinks=True, ignore_errors=False):498        try:499            return self._entry.stat(follow_symlinks=follow_symlinks)500        except OSError:501            if not ignore_errors:502                raise503            return None504 505    def exists(self, *, follow_symlinks=True):506        """Whether this path exists."""507        if not follow_symlinks:508            return True509        return self._stat(ignore_errors=True) is not None510 511    def is_dir(self, *, follow_symlinks=True):512        """Whether this path is a directory."""513        try:514            return self._entry.is_dir(follow_symlinks=follow_symlinks)515        except OSError:516            return False517 518    def is_file(self, *, follow_symlinks=True):519        """Whether this path is a regular file."""520        try:521            return self._entry.is_file(follow_symlinks=follow_symlinks)522        except OSError:523            return False524 525    def is_symlink(self):526        """Whether this path is a symbolic link."""527        try:528            return self._entry.is_symlink()529        except OSError:530            return False531