codekingpro/portable-devtools
114k
1r"""OS routines for NT or Posix depending on what system we're on.2 3This exports:4 - all functions from posix or nt, e.g. unlink, stat, etc.5 - os.path is either posixpath or ntpath6 - os.name is either 'posix' or 'nt'7 - os.curdir is a string representing the current directory (always '.')8 - os.pardir is a string representing the parent directory (always '..')9 - os.sep is the (or a most common) pathname separator ('/' or '\\')10 - os.extsep is the extension separator (always '.')11 - os.altsep is the alternate pathname separator (None or '/')12 - os.pathsep is the component separator used in $PATH etc13 - os.linesep is the line separator in text files ('\n' or '\r\n')14 - os.defpath is the default search path for executables15 - os.devnull is the file path of the null device ('/dev/null', etc.)16 17Programs that import and use 'os' stand a better chance of being18portable between different platforms. Of course, they must then19only use functions that are defined by all platforms (e.g., unlink20and opendir), and leave all pathname manipulation to os.path21(e.g., split and join).22"""23 24#'25import abc26import sys27import stat as st28 29from _collections_abc import _check_methods30 31GenericAlias = type(list[int])32 33_names = sys.builtin_module_names34 35# Note: more names are added to __all__ later.36__all__ = ["altsep", "curdir", "pardir", "sep", "pathsep", "linesep",37 "defpath", "name", "path", "devnull", "SEEK_SET", "SEEK_CUR",38 "SEEK_END", "fsencode", "fsdecode", "get_exec_path", "fdopen",39 "extsep"]40 41def _exists(name):42 return name in globals()43 44def _get_exports_list(module):45 try:46 return list(module.__all__)47 except AttributeError:48 return [n for n in dir(module) if n[0] != '_']49 50# Any new dependencies of the os module and/or changes in path separator51# requires updating importlib as well.52if 'posix' in _names:53 name = 'posix'54 linesep = '\n'55 from posix import *56 try:57 from posix import _exit58 __all__.append('_exit')59 except ImportError:60 pass61 import posixpath as path62 63 try:64 from posix import _have_functions65 except ImportError:66 pass67 try:68 from posix import _create_environ69 except ImportError:70 pass71 72 import posix73 __all__.extend(_get_exports_list(posix))74 del posix75 76elif 'nt' in _names:77 name = 'nt'78 linesep = '\r\n'79 from nt import *80 try:81 from nt import _exit82 __all__.append('_exit')83 except ImportError:84 pass85 import ntpath as path86 87 import nt88 __all__.extend(_get_exports_list(nt))89 del nt90 91 try:92 from nt import _have_functions93 except ImportError:94 pass95 try:96 from nt import _create_environ97 except ImportError:98 pass99 100else:101 raise ImportError('no os specific module found')102 103sys.modules['os.path'] = path104from os.path import (curdir, pardir, sep, pathsep, defpath, extsep, altsep,105 devnull)106 107del _names108 109 110if _exists("_have_functions"):111 _globals = globals()112 def _add(str, fn):113 if (fn in _globals) and (str in _have_functions):114 _set.add(_globals[fn])115 116 _set = set()117 _add("HAVE_FACCESSAT", "access")118 _add("HAVE_FCHMODAT", "chmod")119 _add("HAVE_FCHOWNAT", "chown")120 _add("HAVE_FSTATAT", "stat")121 _add("HAVE_LSTAT", "lstat")122 _add("HAVE_FUTIMESAT", "utime")123 _add("HAVE_LINKAT", "link")124 _add("HAVE_MKDIRAT", "mkdir")125 _add("HAVE_MKFIFOAT", "mkfifo")126 _add("HAVE_MKNODAT", "mknod")127 _add("HAVE_OPENAT", "open")128 _add("HAVE_READLINKAT", "readlink")129 _add("HAVE_RENAMEAT", "rename")130 _add("HAVE_SYMLINKAT", "symlink")131 _add("HAVE_UNLINKAT", "unlink")132 _add("HAVE_UNLINKAT", "rmdir")133 _add("HAVE_UTIMENSAT", "utime")134 supports_dir_fd = _set135 136 _set = set()137 _add("HAVE_FACCESSAT", "access")138 supports_effective_ids = _set139 140 _set = set()141 _add("HAVE_FCHDIR", "chdir")142 _add("HAVE_FCHMOD", "chmod")143 _add("MS_WINDOWS", "chmod")144 _add("HAVE_FCHOWN", "chown")145 _add("HAVE_FDOPENDIR", "listdir")146 _add("HAVE_FDOPENDIR", "scandir")147 _add("HAVE_FEXECVE", "execve")148 _set.add(stat) # fstat always works149 _add("HAVE_FTRUNCATE", "truncate")150 _add("HAVE_FUTIMENS", "utime")151 _add("HAVE_FUTIMES", "utime")152 _add("HAVE_FPATHCONF", "pathconf")153 if _exists("statvfs") and _exists("fstatvfs"): # mac os x10.3154 _add("HAVE_FSTATVFS", "statvfs")155 supports_fd = _set156 157 _set = set()158 _add("HAVE_FACCESSAT", "access")159 # Some platforms don't support lchmod(). Often the function exists160 # anyway, as a stub that always returns ENOSUP or perhaps EOPNOTSUPP.161 # (No, I don't know why that's a good design.) ./configure will detect162 # this and reject it--so HAVE_LCHMOD still won't be defined on such163 # platforms. This is Very Helpful.164 #165 # However, sometimes platforms without a working lchmod() *do* have166 # fchmodat(). (Examples: Linux kernel 3.2 with glibc 2.15,167 # OpenIndiana 3.x.) And fchmodat() has a flag that theoretically makes168 # it behave like lchmod(). So in theory it would be a suitable169 # replacement for lchmod(). But when lchmod() doesn't work, fchmodat()'s170 # flag doesn't work *either*. Sadly ./configure isn't sophisticated171 # enough to detect this condition--it only determines whether or not172 # fchmodat() minimally works.173 #174 # Therefore we simply ignore fchmodat() when deciding whether or not175 # os.chmod supports follow_symlinks. Just checking lchmod() is176 # sufficient. After all--if you have a working fchmodat(), your177 # lchmod() almost certainly works too.178 #179 # _add("HAVE_FCHMODAT", "chmod")180 _add("HAVE_FCHOWNAT", "chown")181 _add("HAVE_FSTATAT", "stat")182 _add("HAVE_LCHFLAGS", "chflags")183 _add("HAVE_LCHMOD", "chmod")184 _add("MS_WINDOWS", "chmod")185 if _exists("lchown"): # mac os x10.3186 _add("HAVE_LCHOWN", "chown")187 _add("HAVE_LINKAT", "link")188 _add("HAVE_LUTIMES", "utime")189 _add("HAVE_LSTAT", "stat")190 _add("HAVE_FSTATAT", "stat")191 _add("HAVE_UTIMENSAT", "utime")192 _add("MS_WINDOWS", "stat")193 supports_follow_symlinks = _set194 195 del _set196 del _have_functions197 del _globals198 del _add199 200 201# Python uses fixed values for the SEEK_ constants; they are mapped202# to native constants if necessary in posixmodule.c203# Other possible SEEK values are directly imported from posixmodule.c204SEEK_SET = 0205SEEK_CUR = 1206SEEK_END = 2207 208# Super directory utilities.209# (Inspired by Eric Raymond; the doc strings are mostly his)210 211def makedirs(name, mode=0o777, exist_ok=False):212 """makedirs(name [, mode=0o777][, exist_ok=False])213 214 Super-mkdir; create a leaf directory and all intermediate ones. Works like215 mkdir, except that any intermediate path segment (not just the rightmost)216 will be created if it does not exist. If the target directory already217 exists, raise an OSError if exist_ok is False. Otherwise no exception is218 raised. This is recursive.219 220 """221 head, tail = path.split(name)222 if not tail:223 head, tail = path.split(head)224 if head and tail and not path.exists(head):225 try:226 makedirs(head, exist_ok=exist_ok)227 except FileExistsError:228 # Defeats race condition when another thread created the path229 pass230 cdir = curdir231 if isinstance(tail, bytes):232 cdir = bytes(curdir, 'ASCII')233 if tail == cdir: # xxx/newdir/. exists if xxx/newdir exists234 return235 try:236 mkdir(name, mode)237 except OSError:238 # Cannot rely on checking for EEXIST, since the operating system239 # could give priority to other errors like EACCES or EROFS240 if not exist_ok or not path.isdir(name):241 raise242 243def removedirs(name):244 """removedirs(name)245 246 Super-rmdir; remove a leaf directory and all empty intermediate247 ones. Works like rmdir except that, if the leaf directory is248 successfully removed, directories corresponding to rightmost path249 segments will be pruned away until either the whole path is250 consumed or an error occurs. Errors during this latter phase are251 ignored -- they generally mean that a directory was not empty.252 253 """254 rmdir(name)255 head, tail = path.split(name)256 if not tail:257 head, tail = path.split(head)258 while head and tail:259 try:260 rmdir(head)261 except OSError:262 break263 head, tail = path.split(head)264 265def renames(old, new):266 """renames(old, new)267 268 Super-rename; create directories as necessary and delete any left269 empty. Works like rename, except creation of any intermediate270 directories needed to make the new pathname good is attempted271 first. After the rename, directories corresponding to rightmost272 path segments of the old name will be pruned until either the273 whole path is consumed or a nonempty directory is found.274 275 Note: this function can fail with the new directory structure made276 if you lack permissions needed to unlink the leaf directory or277 file.278 279 """280 head, tail = path.split(new)281 if head and tail and not path.exists(head):282 makedirs(head)283 rename(old, new)284 head, tail = path.split(old)285 if head and tail:286 try:287 removedirs(head)288 except OSError:289 pass290 291__all__.extend(["makedirs", "removedirs", "renames"])292 293# Private sentinel that makes walk() classify all symlinks and junctions as294# regular files.295_walk_symlinks_as_files = object()296 297def walk(top, topdown=True, onerror=None, followlinks=False):298 """Directory tree generator.299 300 For each directory in the directory tree rooted at top (including top301 itself, but excluding '.' and '..'), yields a 3-tuple302 303 dirpath, dirnames, filenames304 305 dirpath is a string, the path to the directory. dirnames is a list of306 the names of the subdirectories in dirpath (including symlinks to directories,307 and excluding '.' and '..').308 filenames is a list of the names of the non-directory files in dirpath.309 Note that the names in the lists are just names, with no path components.310 To get a full path (which begins with top) to a file or directory in311 dirpath, do os.path.join(dirpath, name).312 313 If optional arg 'topdown' is true or not specified, the triple for a314 directory is generated before the triples for any of its subdirectories315 (directories are generated top down). If topdown is false, the triple316 for a directory is generated after the triples for all of its317 subdirectories (directories are generated bottom up).318 319 When topdown is true, the caller can modify the dirnames list in-place320 (e.g., via del or slice assignment), and walk will only recurse into the321 subdirectories whose names remain in dirnames; this can be used to prune the322 search, or to impose a specific order of visiting. Modifying dirnames when323 topdown is false has no effect on the behavior of os.walk(), since the324 directories in dirnames have already been generated by the time dirnames325 itself is generated. No matter the value of topdown, the list of326 subdirectories is retrieved before the tuples for the directory and its327 subdirectories are generated.328 329 By default errors from the os.scandir() call are ignored. If330 optional arg 'onerror' is specified, it should be a function; it331 will be called with one argument, an OSError instance. It can332 report the error to continue with the walk, or raise the exception333 to abort the walk. Note that the filename is available as the334 filename attribute of the exception object.335 336 By default, os.walk does not follow symbolic links to subdirectories on337 systems that support them. In order to get this functionality, set the338 optional argument 'followlinks' to true.339 340 Caution: if you pass a relative pathname for top, don't change the341 current working directory between resumptions of walk. walk never342 changes the current directory, and assumes that the client doesn't343 either.344 345 Example:346 347 import os348 from os.path import join, getsize349 for root, dirs, files in os.walk('python/Lib/xml'):350 print(root, "consumes ")351 print(sum(getsize(join(root, name)) for name in files), end=" ")352 print("bytes in", len(files), "non-directory files")353 if '__pycache__' in dirs:354 dirs.remove('__pycache__') # don't visit __pycache__ directories355 356 """357 sys.audit("os.walk", top, topdown, onerror, followlinks)358 359 stack = [fspath(top)]360 islink, join = path.islink, path.join361 while stack:362 top = stack.pop()363 if isinstance(top, tuple):364 yield top365 continue366 367 dirs = []368 nondirs = []369 walk_dirs = []370 371 # We may not have read permission for top, in which case we can't372 # get a list of the files the directory contains.373 # We suppress the exception here, rather than blow up for a374 # minor reason when (say) a thousand readable directories are still375 # left to visit.376 try:377 with scandir(top) as entries:378 for entry in entries:379 try:380 if followlinks is _walk_symlinks_as_files:381 is_dir = entry.is_dir(follow_symlinks=False) and not entry.is_junction()382 else:383 is_dir = entry.is_dir()384 except OSError:385 # If is_dir() raises an OSError, consider the entry not to386 # be a directory, same behaviour as os.path.isdir().387 is_dir = False388 389 if is_dir:390 dirs.append(entry.name)391 else:392 nondirs.append(entry.name)393 394 if not topdown and is_dir:395 # Bottom-up: traverse into sub-directory, but exclude396 # symlinks to directories if followlinks is False397 if followlinks:398 walk_into = True399 else:400 try:401 is_symlink = entry.is_symlink()402 except OSError:403 # If is_symlink() raises an OSError, consider the404 # entry not to be a symbolic link, same behaviour405 # as os.path.islink().406 is_symlink = False407 walk_into = not is_symlink408 409 if walk_into:410 walk_dirs.append(entry.path)411 except OSError as error:412 if onerror is not None:413 onerror(error)414 continue415 416 if topdown:417 # Yield before sub-directory traversal if going top down418 yield top, dirs, nondirs419 # Traverse into sub-directories420 for dirname in reversed(dirs):421 new_path = join(top, dirname)422 # bpo-23605: os.path.islink() is used instead of caching423 # entry.is_symlink() result during the loop on os.scandir() because424 # the caller can replace the directory entry during the "yield"425 # above.426 if followlinks or not islink(new_path):427 stack.append(new_path)428 else:429 # Yield after sub-directory traversal if going bottom up430 stack.append((top, dirs, nondirs))431 # Traverse into sub-directories432 for new_path in reversed(walk_dirs):433 stack.append(new_path)434 435__all__.append("walk")436 437if {open, stat} <= supports_dir_fd and {scandir, stat} <= supports_fd:438 439 def fwalk(top=".", topdown=True, onerror=None, *, follow_symlinks=False, dir_fd=None):440 """Directory tree generator.441 442 This behaves exactly like walk(), except that it yields a 4-tuple443 444 dirpath, dirnames, filenames, dirfd445 446 `dirpath`, `dirnames` and `filenames` are identical to walk() output,447 and `dirfd` is a file descriptor referring to the directory `dirpath`.448 449 The advantage of fwalk() over walk() is that it's safe against symlink450 races (when follow_symlinks is False).451 452 If dir_fd is not None, it should be a file descriptor open to a directory,453 and top should be relative; top will then be relative to that directory.454 (dir_fd is always supported for fwalk.)455 456 Caution:457 Since fwalk() yields file descriptors, those are only valid until the458 next iteration step, so you should dup() them if you want to keep them459 for a longer period.460 461 Example:462 463 import os464 for root, dirs, files, rootfd in os.fwalk('python/Lib/xml'):465 print(root, "consumes", end="")466 print(sum(os.stat(name, dir_fd=rootfd).st_size for name in files),467 end="")468 print("bytes in", len(files), "non-directory files")469 if '__pycache__' in dirs:470 dirs.remove('__pycache__') # don't visit __pycache__ directories471 """472 sys.audit("os.fwalk", top, topdown, onerror, follow_symlinks, dir_fd)473 top = fspath(top)474 stack = [(_fwalk_walk, (True, dir_fd, top, top, None))]475 isbytes = isinstance(top, bytes)476 try:477 while stack:478 yield from _fwalk(stack, isbytes, topdown, onerror, follow_symlinks)479 finally:480 # Close any file descriptors still on the stack.481 while stack:482 action, value = stack.pop()483 if action == _fwalk_close:484 close(value)485 486 # Each item in the _fwalk() stack is a pair (action, args).487 _fwalk_walk = 0 # args: (isroot, dirfd, toppath, topname, entry)488 _fwalk_yield = 1 # args: (toppath, dirnames, filenames, topfd)489 _fwalk_close = 2 # args: dirfd490 491 def _fwalk(stack, isbytes, topdown, onerror, follow_symlinks):492 # Note: This uses O(depth of the directory tree) file descriptors: if493 # necessary, it can be adapted to only require O(1) FDs, see issue494 # #13734.495 496 action, value = stack.pop()497 if action == _fwalk_close:498 close(value)499 return500 elif action == _fwalk_yield:501 yield value502 return503 assert action == _fwalk_walk504 isroot, dirfd, toppath, topname, entry = value505 try:506 if not follow_symlinks:507 # Note: To guard against symlink races, we use the standard508 # lstat()/open()/fstat() trick.509 if entry is None:510 orig_st = stat(topname, follow_symlinks=False, dir_fd=dirfd)511 else:512 orig_st = entry.stat(follow_symlinks=False)513 topfd = open(topname, O_RDONLY | O_NONBLOCK, dir_fd=dirfd)514 except OSError as err:515 if isroot:516 raise517 if onerror is not None:518 onerror(err)519 return520 stack.append((_fwalk_close, topfd))521 if not follow_symlinks:522 if isroot and not st.S_ISDIR(orig_st.st_mode):523 return524 if not path.samestat(orig_st, stat(topfd)):525 return526 527 scandir_it = scandir(topfd)528 dirs = []529 nondirs = []530 entries = None if topdown or follow_symlinks else []531 for entry in scandir_it:532 name = entry.name533 if isbytes:534 name = fsencode(name)535 try:536 if entry.is_dir():537 dirs.append(name)538 if entries is not None:539 entries.append(entry)540 else:541 nondirs.append(name)542 except OSError:543 try:544 # Add dangling symlinks, ignore disappeared files545 if entry.is_symlink():546 nondirs.append(name)547 except OSError:548 pass549 550 if topdown:551 yield toppath, dirs, nondirs, topfd552 else:553 stack.append((_fwalk_yield, (toppath, dirs, nondirs, topfd)))554 555 toppath = path.join(toppath, toppath[:0]) # Add trailing slash.556 if entries is None:557 stack.extend(558 (_fwalk_walk, (False, topfd, toppath + name, name, None))559 for name in dirs[::-1])560 else:561 stack.extend(562 (_fwalk_walk, (False, topfd, toppath + name, name, entry))563 for name, entry in zip(dirs[::-1], entries[::-1]))564 565 __all__.append("fwalk")566 567def execl(file, *args):568 """execl(file, *args)569 570 Execute the executable file with argument list args, replacing the571 current process. """572 execv(file, args)573 574def execle(file, *args):575 """execle(file, *args, env)576 577 Execute the executable file with argument list args and578 environment env, replacing the current process. """579 env = args[-1]580 execve(file, args[:-1], env)581 582def execlp(file, *args):583 """execlp(file, *args)584 585 Execute the executable file (which is searched for along $PATH)586 with argument list args, replacing the current process. """587 execvp(file, args)588 589def execlpe(file, *args):590 """execlpe(file, *args, env)591 592 Execute the executable file (which is searched for along $PATH)593 with argument list args and environment env, replacing the current594 process. """595 env = args[-1]596 execvpe(file, args[:-1], env)597 598def execvp(file, args):599 """execvp(file, args)600 601 Execute the executable file (which is searched for along $PATH)602 with argument list args, replacing the current process.603 args may be a list or tuple of strings. """604 _execvpe(file, args)605 606def execvpe(file, args, env):607 """execvpe(file, args, env)608 609 Execute the executable file (which is searched for along $PATH)610 with argument list args and environment env, replacing the611 current process.612 args may be a list or tuple of strings. """613 _execvpe(file, args, env)614 615__all__.extend(["execl","execle","execlp","execlpe","execvp","execvpe"])616 617def _execvpe(file, args, env=None):618 if env is not None:619 exec_func = execve620 argrest = (args, env)621 else:622 exec_func = execv623 argrest = (args,)624 env = environ625 626 if path.dirname(file):627 exec_func(file, *argrest)628 return629 saved_exc = None630 path_list = get_exec_path(env)631 if name != 'nt':632 file = fsencode(file)633 path_list = map(fsencode, path_list)634 for dir in path_list:635 fullname = path.join(dir, file)636 try:637 exec_func(fullname, *argrest)638 except (FileNotFoundError, NotADirectoryError) as e:639 last_exc = e640 except OSError as e:641 last_exc = e642 if saved_exc is None:643 saved_exc = e644 if saved_exc is not None:645 raise saved_exc646 raise last_exc647 648 649def get_exec_path(env=None):650 """Returns the sequence of directories that will be searched for the651 named executable (similar to a shell) when launching a process.652 653 *env* must be an environment variable dict or None. If *env* is None,654 os.environ will be used.655 """656 # Use a local import instead of a global import to limit the number of657 # modules loaded at startup: the os module is always loaded at startup by658 # Python. It may also avoid a bootstrap issue.659 import warnings660 661 if env is None:662 env = environ663 664 # {b'PATH': ...}.get('PATH') and {'PATH': ...}.get(b'PATH') emit a665 # BytesWarning when using python -b or python -bb: ignore the warning666 with warnings.catch_warnings():667 warnings.simplefilter("ignore", BytesWarning)668 669 try:670 path_list = env.get('PATH')671 except TypeError:672 path_list = None673 674 if supports_bytes_environ:675 try:676 path_listb = env[b'PATH']677 except (KeyError, TypeError):678 pass679 else:680 if path_list is not None:681 raise ValueError(682 "env cannot contain 'PATH' and b'PATH' keys")683 path_list = path_listb684 685 if path_list is not None and isinstance(path_list, bytes):686 path_list = fsdecode(path_list)687 688 if path_list is None:689 path_list = defpath690 return path_list.split(pathsep)691 692 693# Change environ to automatically call putenv() and unsetenv()694from _collections_abc import MutableMapping, Mapping695 696class _Environ(MutableMapping):697 def __init__(self, data, encodekey, decodekey, encodevalue, decodevalue):698 self.encodekey = encodekey699 self.decodekey = decodekey700 self.encodevalue = encodevalue701 self.decodevalue = decodevalue702 self._data = data703 704 def __getitem__(self, key):705 try:706 value = self._data[self.encodekey(key)]707 except KeyError:708 # raise KeyError with the original key value709 raise KeyError(key) from None710 return self.decodevalue(value)711 712 def __setitem__(self, key, value):713 key = self.encodekey(key)714 value = self.encodevalue(value)715 putenv(key, value)716 self._data[key] = value717 718 def __delitem__(self, key):719 encodedkey = self.encodekey(key)720 unsetenv(encodedkey)721 try:722 del self._data[encodedkey]723 except KeyError:724 # raise KeyError with the original key value725 raise KeyError(key) from None726 727 def __iter__(self):728 # list() from dict object is an atomic operation729 keys = list(self._data)730 for key in keys:731 yield self.decodekey(key)732 733 def __len__(self):734 return len(self._data)735 736 def __repr__(self):737 formatted_items = ", ".join(738 f"{self.decodekey(key)!r}: {self.decodevalue(value)!r}"739 for key, value in self._data.items()740 )741 return f"environ({{{formatted_items}}})"742 743 def copy(self):744 return dict(self)745 746 def setdefault(self, key, value):747 if key not in self:748 self[key] = value749 return self[key]750 751 def __ior__(self, other):752 self.update(other)753 return self754 755 def __or__(self, other):756 if not isinstance(other, Mapping):757 return NotImplemented758 new = dict(self)759 new.update(other)760 return new761 762 def __ror__(self, other):763 if not isinstance(other, Mapping):764 return NotImplemented765 new = dict(other)766 new.update(self)767 return new768 769def _create_environ_mapping():770 if name == 'nt':771 # Where Env Var Names Must Be UPPERCASE772 def check_str(value):773 if not isinstance(value, str):774 raise TypeError("str expected, not %s" % type(value).__name__)775 return value776 encode = check_str777 decode = str778 def encodekey(key):779 return encode(key).upper()780 data = {}781 for key, value in environ.items():782 data[encodekey(key)] = value783 else:784 # Where Env Var Names Can Be Mixed Case785 encoding = sys.getfilesystemencoding()786 def encode(value):787 if not isinstance(value, str):788 raise TypeError("str expected, not %s" % type(value).__name__)789 return value.encode(encoding, 'surrogateescape')790 def decode(value):791 return value.decode(encoding, 'surrogateescape')792 encodekey = encode793 data = environ794 return _Environ(data,795 encodekey, decode,796 encode, decode)797 798# unicode environ799environ = _create_environ_mapping()800del _create_environ_mapping801 802 803if _exists("_create_environ"):804 def reload_environ():805 data = _create_environ()806 if name == 'nt':807 encodekey = environ.encodekey808 data = {encodekey(key): value809 for key, value in data.items()}810 811 # modify in-place to keep os.environb in sync812 env_data = environ._data813 env_data.clear()814 env_data.update(data)815 816 __all__.append("reload_environ")817 818def getenv(key, default=None):819 """Get an environment variable, return None if it doesn't exist.820 The optional second argument can specify an alternate default.821 key, default and the result are str."""822 return environ.get(key, default)823 824supports_bytes_environ = (name != 'nt')825__all__.extend(("getenv", "supports_bytes_environ"))826 827if supports_bytes_environ:828 def _check_bytes(value):829 if not isinstance(value, bytes):830 raise TypeError("bytes expected, not %s" % type(value).__name__)831 return value832 833 # bytes environ834 environb = _Environ(environ._data,835 _check_bytes, bytes,836 _check_bytes, bytes)837 del _check_bytes838 839 def getenvb(key, default=None):840 """Get an environment variable, return None if it doesn't exist.841 The optional second argument can specify an alternate default.842 key, default and the result are bytes."""843 return environb.get(key, default)844 845 __all__.extend(("environb", "getenvb"))846 847def _fscodec():848 encoding = sys.getfilesystemencoding()849 errors = sys.getfilesystemencodeerrors()850 851 def fsencode(filename):852 """Encode filename (an os.PathLike, bytes, or str) to the filesystem853 encoding with 'surrogateescape' error handler, return bytes unchanged.854 On Windows, use 'strict' error handler if the file system encoding is855 'mbcs' (which is the default encoding).856 """857 filename = fspath(filename) # Does type-checking of `filename`.858 if isinstance(filename, str):859 return filename.encode(encoding, errors)860 else:861 return filename862 863 def fsdecode(filename):864 """Decode filename (an os.PathLike, bytes, or str) from the filesystem865 encoding with 'surrogateescape' error handler, return str unchanged. On866 Windows, use 'strict' error handler if the file system encoding is867 'mbcs' (which is the default encoding).868 """869 filename = fspath(filename) # Does type-checking of `filename`.870 if isinstance(filename, bytes):871 return filename.decode(encoding, errors)872 else:873 return filename874 875 return fsencode, fsdecode876 877fsencode, fsdecode = _fscodec()878del _fscodec879 880# Supply spawn*() (probably only for Unix)881if _exists("fork") and not _exists("spawnv") and _exists("execv"):882 883 P_WAIT = 0884 P_NOWAIT = P_NOWAITO = 1885 886 __all__.extend(["P_WAIT", "P_NOWAIT", "P_NOWAITO"])887 888 # XXX Should we support P_DETACH? I suppose it could fork()**2889 # and close the std I/O streams. Also, P_OVERLAY is the same890 # as execv*()?891 892 def _spawnvef(mode, file, args, env, func):893 # Internal helper; func is the exec*() function to use894 if not isinstance(args, (tuple, list)):895 raise TypeError('argv must be a tuple or a list')896 if not args or not args[0]:897 raise ValueError('argv first element cannot be empty')898 pid = fork()899 if not pid:900 # Child901 try:902 if env is None:903 func(file, args)904 else:905 func(file, args, env)906 except:907 _exit(127)908 else:909 # Parent910 if mode == P_NOWAIT:911 return pid # Caller is responsible for waiting!912 while 1:913 wpid, sts = waitpid(pid, 0)914 if WIFSTOPPED(sts):915 continue916 917 return waitstatus_to_exitcode(sts)918 919 def spawnv(mode, file, args):920 """spawnv(mode, file, args) -> integer921 922Execute file with arguments from args in a subprocess.923If mode == P_NOWAIT return the pid of the process.924If mode == P_WAIT return the process's exit code if it exits normally;925otherwise return -SIG, where SIG is the signal that killed it. """926 return _spawnvef(mode, file, args, None, execv)927 928 def spawnve(mode, file, args, env):929 """spawnve(mode, file, args, env) -> integer930 931Execute file with arguments from args in a subprocess with the932specified environment.933If mode == P_NOWAIT return the pid of the process.934If mode == P_WAIT return the process's exit code if it exits normally;935otherwise return -SIG, where SIG is the signal that killed it. """936 return _spawnvef(mode, file, args, env, execve)937 938 # Note: spawnvp[e] isn't currently supported on Windows939 940 def spawnvp(mode, file, args):941 """spawnvp(mode, file, args) -> integer942 943Execute file (which is looked for along $PATH) with arguments from944args in a subprocess.945If mode == P_NOWAIT return the pid of the process.946If mode == P_WAIT return the process's exit code if it exits normally;947otherwise return -SIG, where SIG is the signal that killed it. """948 return _spawnvef(mode, file, args, None, execvp)949 950 def spawnvpe(mode, file, args, env):951 """spawnvpe(mode, file, args, env) -> integer952 953Execute file (which is looked for along $PATH) with arguments from954args in a subprocess with the supplied environment.955If mode == P_NOWAIT return the pid of the process.956If mode == P_WAIT return the process's exit code if it exits normally;957otherwise return -SIG, where SIG is the signal that killed it. """958 return _spawnvef(mode, file, args, env, execvpe)959 960 961 __all__.extend(["spawnv", "spawnve", "spawnvp", "spawnvpe"])962 963 964if _exists("spawnv"):965 # These aren't supplied by the basic Windows code966 # but can be easily implemented in Python967 968 def spawnl(mode, file, *args):969 """spawnl(mode, file, *args) -> integer970 971Execute file with arguments from args in a subprocess.972If mode == P_NOWAIT return the pid of the process.973If mode == P_WAIT return the process's exit code if it exits normally;974otherwise return -SIG, where SIG is the signal that killed it. """975 return spawnv(mode, file, args)976 977 def spawnle(mode, file, *args):978 """spawnle(mode, file, *args, env) -> integer979 980Execute file with arguments from args in a subprocess with the981supplied environment.982If mode == P_NOWAIT return the pid of the process.983If mode == P_WAIT return the process's exit code if it exits normally;984otherwise return -SIG, where SIG is the signal that killed it. """985 env = args[-1]986 return spawnve(mode, file, args[:-1], env)987 988 989 __all__.extend(["spawnl", "spawnle"])990 991 992if _exists("spawnvp"):993 # At the moment, Windows doesn't implement spawnvp[e],994 # so it won't have spawnlp[e] either.995 def spawnlp(mode, file, *args):996 """spawnlp(mode, file, *args) -> integer997 998Execute file (which is looked for along $PATH) with arguments from999args in a subprocess with the supplied environment.1000If mode == P_NOWAIT return the pid of the process.1001If mode == P_WAIT return the process's exit code if it exits normally;1002otherwise return -SIG, where SIG is the signal that killed it. """1003 return spawnvp(mode, file, args)1004 1005 def spawnlpe(mode, file, *args):1006 """spawnlpe(mode, file, *args, env) -> integer1007 1008Execute file (which is looked for along $PATH) with arguments from1009args in a subprocess with the supplied environment.1010If mode == P_NOWAIT return the pid of the process.1011If mode == P_WAIT return the process's exit code if it exits normally;1012otherwise return -SIG, where SIG is the signal that killed it. """1013 env = args[-1]1014 return spawnvpe(mode, file, args[:-1], env)1015 1016 1017 __all__.extend(["spawnlp", "spawnlpe"])1018 1019# VxWorks has no user space shell provided. As a result, running1020# command in a shell can't be supported.1021if sys.platform != 'vxworks':1022 # Supply os.popen()1023 def popen(cmd, mode="r", buffering=-1):1024 if not isinstance(cmd, str):1025 raise TypeError("invalid cmd type (%s, expected string)" % type(cmd))1026 if mode not in ("r", "w"):1027 raise ValueError("invalid mode %r" % mode)1028 if buffering == 0 or buffering is None:1029 raise ValueError("popen() does not support unbuffered streams")1030 import subprocess1031 if mode == "r":1032 proc = subprocess.Popen(cmd,1033 shell=True, text=True,1034 stdout=subprocess.PIPE,1035 bufsize=buffering)1036 return _wrap_close(proc.stdout, proc)1037 else:1038 proc = subprocess.Popen(cmd,1039 shell=True, text=True,1040 stdin=subprocess.PIPE,1041 bufsize=buffering)1042 return _wrap_close(proc.stdin, proc)1043 1044 # Helper for popen() -- a proxy for a file whose close waits for the process1045 class _wrap_close:1046 def __init__(self, stream, proc):1047 self._stream = stream1048 self._proc = proc1049 def close(self):1050 self._stream.close()1051 returncode = self._proc.wait()1052 if returncode == 0:1053 return None1054 if name == 'nt':1055 return returncode1056 else:1057 return returncode << 8 # Shift left to match old behavior1058 def __enter__(self):1059 return self1060 def __exit__(self, *args):1061 self.close()1062 def __getattr__(self, name):1063 return getattr(self._stream, name)1064 def __iter__(self):1065 return iter(self._stream)1066 1067 __all__.append("popen")1068 1069# Supply os.fdopen()1070def fdopen(fd, mode="r", buffering=-1, encoding=None, *args, **kwargs):1071 if not isinstance(fd, int):1072 raise TypeError("invalid fd type (%s, expected integer)" % type(fd))1073 import io1074 if "b" not in mode:1075 encoding = io.text_encoding(encoding)1076 return io.open(fd, mode, buffering, encoding, *args, **kwargs)1077 1078 1079# For testing purposes, make sure the function is available when the C1080# implementation exists.1081def _fspath(path):1082 """Return the path representation of a path-like object.1083 1084 If str or bytes is passed in, it is returned unchanged. Otherwise the1085 os.PathLike interface is used to get the path representation. If the1086 path representation is not str or bytes, TypeError is raised. If the1087 provided path is not str, bytes, or os.PathLike, TypeError is raised.1088 """1089 if isinstance(path, (str, bytes)):1090 return path1091 1092 # Work from the object's type to match method resolution of other magic1093 # methods.1094 path_type = type(path)1095 try:1096 path_repr = path_type.__fspath__(path)1097 except AttributeError:1098 if hasattr(path_type, '__fspath__'):1099 raise1100 else:1101 raise TypeError("expected str, bytes or os.PathLike object, "1102 "not " + path_type.__name__)1103 except TypeError:1104 if path_type.__fspath__ is None:1105 raise TypeError("expected str, bytes or os.PathLike object, "1106 "not " + path_type.__name__) from None1107 else:1108 raise1109 if isinstance(path_repr, (str, bytes)):1110 return path_repr1111 else:1112 raise TypeError("expected {}.__fspath__() to return str or bytes, "1113 "not {}".format(path_type.__name__,1114 type(path_repr).__name__))1115 1116# If there is no C implementation, make the pure Python version the1117# implementation as transparently as possible.1118if not _exists('fspath'):1119 fspath = _fspath1120 fspath.__name__ = "fspath"1121 1122 1123class PathLike(abc.ABC):1124 1125 """Abstract base class for implementing the file system path protocol."""1126 1127 __slots__ = ()1128 1129 @abc.abstractmethod1130 def __fspath__(self):1131 """Return the file system path representation of the object."""1132 raise NotImplementedError1133 1134 @classmethod1135 def __subclasshook__(cls, subclass):1136 if cls is PathLike:1137 return _check_methods(subclass, '__fspath__')1138 return NotImplemented1139 1140 __class_getitem__ = classmethod(GenericAlias)1141 1142 1143if name == 'nt':1144 class _AddedDllDirectory:1145 def __init__(self, path, cookie, remove_dll_directory):1146 self.path = path1147 self._cookie = cookie1148 self._remove_dll_directory = remove_dll_directory1149 def close(self):1150 self._remove_dll_directory(self._cookie)1151 self.path = None1152 def __enter__(self):1153 return self1154 def __exit__(self, *args):1155 self.close()1156 def __repr__(self):1157 if self.path:1158 return "<AddedDllDirectory({!r})>".format(self.path)1159 return "<AddedDllDirectory()>"1160 1161 def add_dll_directory(path):1162 """Add a path to the DLL search path.1163 1164 This search path is used when resolving dependencies for imported1165 extension modules (the module itself is resolved through sys.path),1166 and also by ctypes.1167 1168 Remove the directory by calling close() on the returned object or1169 using it in a with statement.1170 """1171 import nt1172 cookie = nt._add_dll_directory(path)1173 return _AddedDllDirectory(1174 path,1175 cookie,1176 nt._remove_dll_directory1177 )1178 1179 1180if _exists('sched_getaffinity') and sys._get_cpu_count_config() < 0:1181 def process_cpu_count():1182 """1183 Get the number of CPUs of the current process.1184 1185 Return the number of logical CPUs usable by the calling thread of the1186 current process. Return None if indeterminable.1187 """1188 return len(sched_getaffinity(0))1189else:1190 # Just an alias to cpu_count() (same docstring)1191 process_cpu_count = cpu_count1192