Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import errno
4import getpass
5import hashlib
6import logging
7import os
8import pathlib
9import posixpath
10import shutil
11import stat
12import sys
13import sysconfig
14import urllib.parse
15from collections.abc import Callable, Generator, Iterable, Iterator, Mapping, Sequence
16from dataclasses import dataclass
17from functools import partial
18from io import StringIO
19from itertools import filterfalse, tee, zip_longest
20from pathlib import Path
21from types import FunctionType, TracebackType
22from typing import (
23 Any,
24 BinaryIO,
25 TextIO,
26 TypeVar,
27 cast,
28)
30from pip._vendor.packaging.requirements import Requirement
31from pip._vendor.pyproject_hooks import BuildBackendHookCaller
33from pip import __file__ as pip_location
34from pip import __version__
35from pip._internal.exceptions import CommandError, ExternallyManagedEnvironment
36from pip._internal.locations import get_major_minor_version
37from pip._internal.utils.compat import WINDOWS
38from pip._internal.utils.retry import retry
39from pip._internal.utils.virtualenv import running_under_virtualenv
41__all__ = [
42 "rmtree",
43 "display_path",
44 "backup_dir",
45 "ask",
46 "splitext",
47 "format_size",
48 "is_installable_dir",
49 "normalize_path",
50 "renames",
51 "get_prog",
52 "ensure_dir",
53 "remove_auth_from_url",
54 "check_externally_managed",
55 "ConfiguredBuildBackendHookCaller",
56]
58logger = logging.getLogger(__name__)
60T = TypeVar("T")
61ExcInfo = tuple[type[BaseException], BaseException, TracebackType]
62VersionInfo = tuple[int, int, int]
63NetlocTuple = tuple[str, tuple[str | None, str | None]]
64OnExc = Callable[[FunctionType, Path, BaseException], Any]
65OnErr = Callable[[FunctionType, Path, ExcInfo], Any]
67FILE_CHUNK_SIZE = 1024 * 1024
70def get_pip_version() -> str:
71 pip_pkg_dir = os.path.join(os.path.dirname(__file__), "..", "..")
72 pip_pkg_dir = os.path.abspath(pip_pkg_dir)
74 return f"pip {__version__} from {pip_pkg_dir} (python {get_major_minor_version()})"
77def get_runnable_pip() -> str:
78 """Get a file to pass to a Python executable, to run the currently-running pip.
80 This is used to run a pip subprocess, for installing requirements into the build
81 environment.
82 """
83 source = pathlib.Path(pip_location).resolve().parent
85 if not source.is_dir():
86 # This would happen if someone is using pip from inside a zip file. In that
87 # case, we can use that directly.
88 return str(source)
90 return os.fsdecode(source / "__pip-runner__.py")
93def normalize_version_info(py_version_info: tuple[int, ...]) -> tuple[int, int, int]:
94 """
95 Convert a tuple of ints representing a Python version to one of length
96 three.
98 :param py_version_info: a tuple of ints representing a Python version,
99 or None to specify no version. The tuple can have any length.
101 :return: a tuple of length three if `py_version_info` is non-None.
102 Otherwise, return `py_version_info` unchanged (i.e. None).
103 """
104 if len(py_version_info) < 3:
105 py_version_info += (3 - len(py_version_info)) * (0,)
106 elif len(py_version_info) > 3:
107 py_version_info = py_version_info[:3]
109 return cast("VersionInfo", py_version_info)
112def ensure_dir(path: str) -> None:
113 """os.path.makedirs without EEXIST."""
114 try:
115 os.makedirs(path)
116 except OSError as e:
117 # Windows can raise spurious ENOTEMPTY errors. See #6426.
118 if e.errno != errno.EEXIST and e.errno != errno.ENOTEMPTY:
119 raise
122def get_prog() -> str:
123 try:
124 prog = os.path.basename(sys.argv[0])
125 if prog in ("__main__.py", "-c"):
126 return f"{sys.executable} -m pip"
127 else:
128 return prog
129 except (AttributeError, TypeError, IndexError):
130 pass
131 return "pip"
134# Retry every half second for up to 3 seconds
135@retry(stop_after_delay=3, wait=0.5)
136def rmtree(dir: str, ignore_errors: bool = False, onexc: OnExc | None = None) -> None:
137 if ignore_errors:
138 onexc = _onerror_ignore
139 if onexc is None:
140 onexc = _onerror_reraise
141 handler: OnErr = partial(rmtree_errorhandler, onexc=onexc)
142 if sys.version_info >= (3, 12):
143 # See https://docs.python.org/3.12/whatsnew/3.12.html#shutil.
144 shutil.rmtree(dir, onexc=handler) # type: ignore
145 else:
146 shutil.rmtree(dir, onerror=handler) # type: ignore
149def _onerror_ignore(*_args: Any) -> None:
150 pass
153def _onerror_reraise(*_args: Any) -> None:
154 raise # noqa: PLE0704 - Bare exception used to reraise existing exception
157def rmtree_errorhandler(
158 func: FunctionType,
159 path: Path,
160 exc_info: ExcInfo | BaseException,
161 *,
162 onexc: OnExc = _onerror_reraise,
163) -> None:
164 """
165 `rmtree` error handler to 'force' a file remove (i.e. like `rm -f`).
167 * If a file is readonly then it's write flag is set and operation is
168 retried.
170 * `onerror` is the original callback from `rmtree(... onerror=onerror)`
171 that is chained at the end if the "rm -f" still fails.
172 """
173 try:
174 st_mode = os.stat(path).st_mode
175 except OSError:
176 # it's equivalent to os.path.exists
177 return
179 if not st_mode & stat.S_IWRITE:
180 # convert to read/write
181 try:
182 os.chmod(path, st_mode | stat.S_IWRITE)
183 except OSError:
184 pass
185 else:
186 # use the original function to repeat the operation
187 try:
188 func(path)
189 return
190 except OSError:
191 pass
193 if not isinstance(exc_info, BaseException):
194 _, exc_info, _ = exc_info
195 onexc(func, path, exc_info)
198def display_path(path: str) -> str:
199 """Gives the display value for a given path, making it relative to cwd
200 if possible."""
201 try:
202 relative = Path(path).relative_to(Path.cwd())
203 except ValueError:
204 # If the path isn't relative to the CWD, leave it alone
205 return path
206 return os.path.join(".", relative)
209def backup_dir(dir: str, ext: str = ".bak") -> str:
210 """Figure out the name of a directory to back up the given dir to
211 (adding .bak, .bak2, etc)"""
212 n = 1
213 extension = ext
214 while os.path.exists(dir + extension):
215 n += 1
216 extension = ext + str(n)
217 return dir + extension
220def ask_path_exists(message: str, options: Iterable[str]) -> str:
221 for action in os.environ.get("PIP_EXISTS_ACTION", "").split():
222 if action in options:
223 return action
224 return ask(message, options)
227def _check_no_input(message: str) -> None:
228 """Raise an error if no input is allowed."""
229 if os.environ.get("PIP_NO_INPUT"):
230 raise Exception(
231 f"No input was expected ($PIP_NO_INPUT set); question: {message}"
232 )
235def ask(message: str, options: Iterable[str]) -> str:
236 """Ask the message interactively, with the given possible responses"""
237 while 1:
238 _check_no_input(message)
239 response = input(message)
240 response = response.strip().lower()
241 if response not in options:
242 print(
243 "Your response ({!r}) was not one of the expected responses: "
244 "{}".format(response, ", ".join(options))
245 )
246 else:
247 return response
250def ask_input(message: str) -> str:
251 """Ask for input interactively."""
252 _check_no_input(message)
253 return input(message)
256def ask_password(message: str) -> str:
257 """Ask for a password interactively."""
258 _check_no_input(message)
259 return getpass.getpass(message)
262def strtobool(val: str) -> int:
263 """Convert a string representation of truth to true (1) or false (0).
265 True values are 'y', 'yes', 't', 'true', 'on', and '1'; false values
266 are 'n', 'no', 'f', 'false', 'off', and '0'. Raises ValueError if
267 'val' is anything else.
268 """
269 val = val.lower()
270 if val in ("y", "yes", "t", "true", "on", "1"):
271 return 1
272 elif val in ("n", "no", "f", "false", "off", "0"):
273 return 0
274 else:
275 raise ValueError(f"invalid truth value {val!r}")
278def format_size(bytes: float) -> str:
279 if bytes > 1000 * 1000:
280 return f"{bytes / 1000.0 / 1000:.1f} MB"
281 elif bytes > 10 * 1000:
282 return f"{int(bytes / 1000)} kB"
283 elif bytes > 1000:
284 return f"{bytes / 1000.0:.1f} kB"
285 else:
286 return f"{int(bytes)} bytes"
289def tabulate(rows: Iterable[Iterable[Any]]) -> tuple[list[str], list[int]]:
290 """Return a list of formatted rows and a list of column sizes.
292 For example::
294 >>> tabulate([['foobar', 2000], [0xdeadbeef]])
295 (['foobar 2000', '3735928559'], [10, 4])
296 """
297 rows = [tuple(map(str, row)) for row in rows]
298 sizes = [max(map(len, col)) for col in zip_longest(*rows, fillvalue="")]
299 table = [" ".join(map(str.ljust, row, sizes)).rstrip() for row in rows]
300 return table, sizes
303def is_installable_dir(path: str) -> bool:
304 """Is path is a directory containing pyproject.toml or setup.py?
306 If pyproject.toml exists, this is a PEP 517 project. Otherwise we look for
307 a legacy setuptools layout by identifying setup.py. We don't check for the
308 setup.cfg because using it without setup.py is only available for PEP 517
309 projects, which are already covered by the pyproject.toml check.
310 """
311 if not os.path.isdir(path):
312 return False
313 if os.path.isfile(os.path.join(path, "pyproject.toml")):
314 return True
315 if os.path.isfile(os.path.join(path, "setup.py")):
316 return True
317 return False
320def read_chunks(
321 file: BinaryIO, size: int = FILE_CHUNK_SIZE
322) -> Generator[bytes, None, None]:
323 """Yield pieces of data from a file-like object until EOF."""
324 while True:
325 chunk = file.read(size)
326 if not chunk:
327 break
328 yield chunk
331def normalize_path(path: str, resolve_symlinks: bool = True) -> str:
332 """
333 Convert a path to its canonical, case-normalized, absolute version.
335 """
336 path = os.path.expanduser(path)
337 if resolve_symlinks:
338 path = os.path.realpath(path)
339 else:
340 path = os.path.abspath(path)
341 return os.path.normcase(path)
344def splitext(path: str) -> tuple[str, str]:
345 """Like os.path.splitext, but take off .tar too"""
346 base, ext = posixpath.splitext(path)
347 if base.lower().endswith(".tar"):
348 ext = base[-4:] + ext
349 base = base[:-4]
350 return base, ext
353def renames(old: str, new: str) -> None:
354 """Like os.renames(), but handles renaming across devices."""
355 # Implementation borrowed from os.renames().
356 head, tail = os.path.split(new)
357 if head and tail and not os.path.exists(head):
358 os.makedirs(head)
360 shutil.move(old, new)
362 head, tail = os.path.split(old)
363 if head and tail:
364 try:
365 os.removedirs(head)
366 except OSError:
367 pass
370def is_local(path: str) -> bool:
371 """
372 Return True if path is within sys.prefix, if we're running in a virtualenv.
374 If we're not in a virtualenv, all paths are considered "local."
376 Caution: this function assumes the head of path has been normalized
377 with normalize_path.
378 """
379 if not running_under_virtualenv():
380 return True
381 return path.startswith(normalize_path(sys.prefix))
384def write_output(msg: Any, *args: Any) -> None:
385 logger.info(msg, *args)
388class StreamWrapper(StringIO):
389 orig_stream: TextIO
391 @classmethod
392 def from_stream(cls, orig_stream: TextIO) -> StreamWrapper:
393 ret = cls()
394 ret.orig_stream = orig_stream
395 return ret
397 # compileall.compile_dir() needs stdout.encoding to print to stdout
398 # type ignore is because TextIOBase.encoding is writeable
399 @property
400 def encoding(self) -> str: # type: ignore
401 return self.orig_stream.encoding
404# Simulates an enum
405def enum(*sequential: Any, **named: Any) -> type[Any]:
406 enums = dict(zip(sequential, range(len(sequential))), **named)
407 reverse = {value: key for key, value in enums.items()}
408 enums["reverse_mapping"] = reverse
409 return type("Enum", (), enums)
412def build_netloc(host: str, port: int | None) -> str:
413 """
414 Build a netloc from a host-port pair
415 """
416 if port is None:
417 return host
418 if ":" in host:
419 # Only wrap host with square brackets when it is IPv6
420 host = f"[{host}]"
421 return f"{host}:{port}"
424def build_url_from_netloc(netloc: str, scheme: str = "https") -> str:
425 """
426 Build a full URL from a netloc.
427 """
428 if netloc.count(":") >= 2 and "@" not in netloc and "[" not in netloc:
429 # It must be a bare IPv6 address, so wrap it with brackets.
430 netloc = f"[{netloc}]"
431 return f"{scheme}://{netloc}"
434def parse_netloc(netloc: str) -> tuple[str | None, int | None]:
435 """
436 Return the host-port pair from a netloc.
437 """
438 url = build_url_from_netloc(netloc)
439 parsed = urllib.parse.urlparse(url)
440 return parsed.hostname, parsed.port
443def split_auth_from_netloc(netloc: str) -> NetlocTuple:
444 """
445 Parse out and remove the auth information from a netloc.
447 Returns: (netloc, (username, password)).
448 """
449 if "@" not in netloc:
450 return netloc, (None, None)
452 # Split from the right because that's how urllib.parse.urlsplit()
453 # behaves if more than one @ is present (which can be checked using
454 # the password attribute of urlsplit()'s return value).
455 auth, netloc = netloc.rsplit("@", 1)
456 pw: str | None = None
457 if ":" in auth:
458 # Split from the left because that's how urllib.parse.urlsplit()
459 # behaves if more than one : is present (which again can be checked
460 # using the password attribute of the return value)
461 user, pw = auth.split(":", 1)
462 else:
463 user, pw = auth, None
465 user = urllib.parse.unquote(user)
466 if pw is not None:
467 pw = urllib.parse.unquote(pw)
469 return netloc, (user, pw)
472def redact_netloc(netloc: str) -> str:
473 """
474 Replace the sensitive data in a netloc with "****", if it exists.
476 For example:
477 - "user:pass@example.com" returns "user:****@example.com"
478 - "accesstoken@example.com" returns "****@example.com"
479 """
480 netloc, (user, password) = split_auth_from_netloc(netloc)
481 if user is None:
482 return netloc
483 if password is None:
484 user = "****"
485 password = ""
486 else:
487 user = urllib.parse.quote(user)
488 password = ":****"
489 return f"{user}{password}@{netloc}"
492def _transform_url(
493 url: str, transform_netloc: Callable[[str], tuple[Any, ...]]
494) -> tuple[str, NetlocTuple]:
495 """Transform and replace netloc in a url.
497 transform_netloc is a function taking the netloc and returning a
498 tuple. The first element of this tuple is the new netloc. The
499 entire tuple is returned.
501 Returns a tuple containing the transformed url as item 0 and the
502 original tuple returned by transform_netloc as item 1.
503 """
504 purl = urllib.parse.urlsplit(url)
505 netloc_tuple = transform_netloc(purl.netloc)
506 # stripped url
507 url_pieces = (purl.scheme, netloc_tuple[0], purl.path, purl.query, purl.fragment)
508 surl = urllib.parse.urlunsplit(url_pieces)
509 return surl, cast("NetlocTuple", netloc_tuple)
512def _get_netloc(netloc: str) -> NetlocTuple:
513 return split_auth_from_netloc(netloc)
516def _redact_netloc(netloc: str) -> tuple[str]:
517 return (redact_netloc(netloc),)
520def split_auth_netloc_from_url(
521 url: str,
522) -> tuple[str, str, tuple[str | None, str | None]]:
523 """
524 Parse a url into separate netloc, auth, and url with no auth.
526 Returns: (url_without_auth, netloc, (username, password))
527 """
528 url_without_auth, (netloc, auth) = _transform_url(url, _get_netloc)
529 return url_without_auth, netloc, auth
532def remove_auth_from_url(url: str) -> str:
533 """Return a copy of url with 'username:password@' removed."""
534 # username/pass params are passed to subversion through flags
535 # and are not recognized in the url.
536 return _transform_url(url, _get_netloc)[0]
539def redact_auth_from_url(url: str) -> str:
540 """Replace the password in a given url with ****."""
541 return _transform_url(url, _redact_netloc)[0]
544def redact_auth_from_requirement(req: Requirement) -> str:
545 """Replace the password in a given requirement url with ****."""
546 if not req.url:
547 return str(req)
548 return str(req).replace(req.url, redact_auth_from_url(req.url))
551@dataclass(frozen=True)
552class HiddenText:
553 secret: str
554 redacted: str
556 def __repr__(self) -> str:
557 return f"<HiddenText {str(self)!r}>"
559 def __str__(self) -> str:
560 return self.redacted
562 def __eq__(self, other: object) -> bool:
563 # Equality is particularly useful for testing.
564 if type(self) is type(other):
565 # The string being used for redaction doesn't also have to match,
566 # just the raw, original string.
567 return self.secret == other.secret
568 return NotImplemented
570 # Disable hashing, since we have a custom __eq__ and don't need hash-ability
571 # (yet). The only required property of hashing is that objects which compare
572 # equal have the same hash value.
573 __hash__ = None # type: ignore[assignment]
576def hide_value(value: str) -> HiddenText:
577 return HiddenText(value, redacted="****")
580def hide_url(url: str) -> HiddenText:
581 redacted = redact_auth_from_url(url)
582 return HiddenText(url, redacted=redacted)
585def protect_pip_from_modification_on_windows(modifying_pip: bool) -> None:
586 """Protection of pip.exe from modification on Windows
588 On Windows, any operation modifying pip should be run as:
589 python -m pip ...
590 """
591 pip_names = [
592 "pip",
593 f"pip{sys.version_info.major}",
594 f"pip{sys.version_info.major}.{sys.version_info.minor}",
595 ]
597 # See https://github.com/pypa/pip/issues/1299 for more discussion
598 should_show_use_python_msg = (
599 modifying_pip and WINDOWS and os.path.basename(sys.argv[0]) in pip_names
600 )
602 if should_show_use_python_msg:
603 new_command = [sys.executable, "-m", "pip"] + sys.argv[1:]
604 raise CommandError(
605 "To modify pip, please run the following command:\n{}".format(
606 " ".join(new_command)
607 )
608 )
611def check_externally_managed() -> None:
612 """Check whether the current environment is externally managed.
614 If the ``EXTERNALLY-MANAGED`` config file is found, the current environment
615 is considered externally managed, and an ExternallyManagedEnvironment is
616 raised.
617 """
618 if running_under_virtualenv():
619 return
620 marker = os.path.join(sysconfig.get_path("stdlib"), "EXTERNALLY-MANAGED")
621 if not os.path.isfile(marker):
622 return
623 raise ExternallyManagedEnvironment.from_config(marker)
626def is_console_interactive() -> bool:
627 """Is this console interactive?"""
628 return sys.stdin is not None and sys.stdin.isatty()
631def hash_file(path: str, blocksize: int = 1 << 20) -> tuple[Any, int]:
632 """Return (hash, length) for path using hashlib.sha256()"""
634 h = hashlib.sha256()
635 length = 0
636 with open(path, "rb") as f:
637 for block in read_chunks(f, size=blocksize):
638 length += len(block)
639 h.update(block)
640 return h, length
643def pairwise(iterable: Iterable[Any]) -> Iterator[tuple[Any, Any]]:
644 """
645 Return paired elements.
647 For example:
648 s -> (s0, s1), (s2, s3), (s4, s5), ...
649 """
650 iterable = iter(iterable)
651 return zip_longest(iterable, iterable)
654def partition(
655 pred: Callable[[T], bool], iterable: Iterable[T]
656) -> tuple[Iterable[T], Iterable[T]]:
657 """
658 Use a predicate to partition entries into false entries and true entries,
659 like
661 partition(is_odd, range(10)) --> 0 2 4 6 8 and 1 3 5 7 9
662 """
663 t1, t2 = tee(iterable)
664 return filterfalse(pred, t1), filter(pred, t2)
667class ConfiguredBuildBackendHookCaller(BuildBackendHookCaller):
668 def __init__(
669 self,
670 config_holder: Any,
671 source_dir: str,
672 build_backend: str,
673 backend_path: str | None = None,
674 runner: Callable[..., None] | None = None,
675 python_executable: str | None = None,
676 ):
677 super().__init__(
678 source_dir, build_backend, backend_path, runner, python_executable
679 )
680 self.config_holder = config_holder
682 def build_wheel(
683 self,
684 wheel_directory: str,
685 config_settings: Mapping[str, Any] | None = None,
686 metadata_directory: str | None = None,
687 ) -> str:
688 cs = self.config_holder.config_settings
689 return super().build_wheel(
690 wheel_directory, config_settings=cs, metadata_directory=metadata_directory
691 )
693 def build_sdist(
694 self,
695 sdist_directory: str,
696 config_settings: Mapping[str, Any] | None = None,
697 ) -> str:
698 cs = self.config_holder.config_settings
699 return super().build_sdist(sdist_directory, config_settings=cs)
701 def build_editable(
702 self,
703 wheel_directory: str,
704 config_settings: Mapping[str, Any] | None = None,
705 metadata_directory: str | None = None,
706 ) -> str:
707 cs = self.config_holder.config_settings
708 return super().build_editable(
709 wheel_directory, config_settings=cs, metadata_directory=metadata_directory
710 )
712 def get_requires_for_build_wheel(
713 self, config_settings: Mapping[str, Any] | None = None
714 ) -> Sequence[str]:
715 cs = self.config_holder.config_settings
716 return super().get_requires_for_build_wheel(config_settings=cs)
718 def get_requires_for_build_sdist(
719 self, config_settings: Mapping[str, Any] | None = None
720 ) -> Sequence[str]:
721 cs = self.config_holder.config_settings
722 return super().get_requires_for_build_sdist(config_settings=cs)
724 def get_requires_for_build_editable(
725 self, config_settings: Mapping[str, Any] | None = None
726 ) -> Sequence[str]:
727 cs = self.config_holder.config_settings
728 return super().get_requires_for_build_editable(config_settings=cs)
730 def prepare_metadata_for_build_wheel(
731 self,
732 metadata_directory: str,
733 config_settings: Mapping[str, Any] | None = None,
734 _allow_fallback: bool = True,
735 ) -> str:
736 cs = self.config_holder.config_settings
737 return super().prepare_metadata_for_build_wheel(
738 metadata_directory=metadata_directory,
739 config_settings=cs,
740 _allow_fallback=_allow_fallback,
741 )
743 def prepare_metadata_for_build_editable(
744 self,
745 metadata_directory: str,
746 config_settings: Mapping[str, Any] | None = None,
747 _allow_fallback: bool = True,
748 ) -> str | None:
749 cs = self.config_holder.config_settings
750 return super().prepare_metadata_for_build_editable(
751 metadata_directory=metadata_directory,
752 config_settings=cs,
753 _allow_fallback=_allow_fallback,
754 )
757def warn_if_run_as_root() -> None:
758 """Output a warning for sudo users on Unix.
760 In a virtual environment, sudo pip still writes to virtualenv.
761 On Windows, users may run pip as Administrator without issues.
762 This warning only applies to Unix root users outside of virtualenv.
763 """
764 if running_under_virtualenv():
765 return
766 if not hasattr(os, "getuid"):
767 return
768 # On Windows, there are no "system managed" Python packages. Installing as
769 # Administrator via pip is the correct way of updating system environments.
770 #
771 # We choose sys.platform over utils.compat.WINDOWS here to enable Mypy platform
772 # checks: https://mypy.readthedocs.io/en/stable/common_issues.html
773 if sys.platform == "win32" or sys.platform == "cygwin":
774 return
776 if os.getuid() != 0:
777 return
779 logger.warning(
780 "Running pip as the 'root' user can result in broken permissions and "
781 "conflicting behaviour with the system package manager, possibly "
782 "rendering your system unusable. "
783 "It is recommended to use a virtual environment instead: "
784 "https://pip.pypa.io/warnings/venv. "
785 "Use the --root-user-action option if you know what you are doing and "
786 "want to suppress this warning."
787 )