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 functools
4import os
5import sys
6import sysconfig
7from collections.abc import Callable, Generator, Iterable
8from importlib.util import cache_from_source
9from typing import Any
11from pip._internal.exceptions import LegacyDistutilsInstall, UninstallMissingRecord
12from pip._internal.locations import get_bin_prefix, get_bin_user
13from pip._internal.metadata import BaseDistribution
14from pip._internal.utils.compat import WINDOWS
15from pip._internal.utils.egg_link import egg_link_path_from_location
16from pip._internal.utils.logging import getLogger, indent_log
17from pip._internal.utils.misc import ask, normalize_path, renames, rmtree
18from pip._internal.utils.temp_dir import AdjacentTempDirectory, TempDirectory
19from pip._internal.utils.virtualenv import running_under_virtualenv
21logger = getLogger(__name__)
24def _script_names(
25 bin_dir: str, script_name: str, is_gui: bool
26) -> Generator[str, None, None]:
27 """Create the fully qualified name of the files created by
28 {console,gui}_scripts for the given ``dist``.
29 Returns the list of file names
30 """
31 exe_name = os.path.join(bin_dir, script_name)
32 yield exe_name
33 if not WINDOWS:
34 return
35 yield f"{exe_name}.exe"
36 yield f"{exe_name}.exe.manifest"
37 if is_gui:
38 yield f"{exe_name}-script.pyw"
39 else:
40 yield f"{exe_name}-script.py"
43def _unique(
44 fn: Callable[..., Generator[Any, None, None]],
45) -> Callable[..., Generator[Any, None, None]]:
46 @functools.wraps(fn)
47 def unique(*args: Any, **kw: Any) -> Generator[Any, None, None]:
48 seen: set[Any] = set()
49 for item in fn(*args, **kw):
50 if item not in seen:
51 seen.add(item)
52 yield item
54 return unique
57@_unique
58def uninstallation_paths(dist: BaseDistribution) -> Generator[str, None, None]:
59 """
60 Yield all the uninstallation paths for dist based on RECORD-without-.py[co]
62 Yield paths to all the files in RECORD. For each .py file in RECORD, add
63 the .pyc and .pyo in the same directory.
65 UninstallPathSet.add() takes care of the __pycache__ .py[co].
67 If RECORD is not found, raises an error,
68 with possible information from the INSTALLER file.
70 https://packaging.python.org/specifications/recording-installed-packages/
71 """
72 location = dist.location
73 assert location is not None, "not installed"
75 entries = dist.iter_declared_entries()
76 if entries is None:
77 raise UninstallMissingRecord(distribution=dist)
79 for entry in entries:
80 path = os.path.join(location, entry)
81 yield path
82 if path.endswith(".py"):
83 dn, fn = os.path.split(path)
84 base = fn[:-3]
85 path = os.path.join(dn, base + ".pyc")
86 yield path
87 path = os.path.join(dn, base + ".pyo")
88 yield path
91def compact(paths: Iterable[str]) -> set[str]:
92 """Compact a path set to contain the minimal number of paths
93 necessary to contain all paths in the set. If /a/path/ and
94 /a/path/to/a/file.txt are both in the set, leave only the
95 shorter path."""
97 sep = os.path.sep
98 short_paths: set[str] = set()
99 prefixes: set[str] = set()
101 for path in sorted(paths, key=len):
102 current = path[:-1] if path.endswith(sep + "*") else path
103 parent = current.rstrip(sep)
104 while parent:
105 if parent in prefixes:
106 break
107 next_parent = os.path.dirname(parent).rstrip(sep)
108 parent = "" if next_parent == parent else next_parent
109 else:
110 short_paths.add(path)
111 prefixes.add(current.rstrip(sep))
112 return short_paths
115def compress_for_rename(paths: Iterable[str]) -> set[str]:
116 """Returns a set containing the paths that need to be renamed.
118 This set may include directories when the original sequence of paths
119 included every file on disk.
120 """
121 case_map = {os.path.normcase(p): p for p in paths}
122 remaining = set(case_map)
123 unchecked = sorted({os.path.split(p)[0] for p in case_map.values()}, key=len)
124 wildcards: set[str] = set()
126 def norm_join(*a: str) -> str:
127 return os.path.normcase(os.path.join(*a))
129 for root in unchecked:
130 if any(os.path.normcase(root).startswith(w) for w in wildcards):
131 # This directory has already been handled.
132 continue
134 all_files: set[str] = set()
135 all_subdirs: set[str] = set()
136 for dirname, subdirs, files in os.walk(root):
137 all_subdirs.update(norm_join(root, dirname, d) for d in subdirs)
138 all_files.update(norm_join(root, dirname, f) for f in files)
139 # If all the files we found are in our remaining set of files to
140 # remove, then remove them from the latter set and add a wildcard
141 # for the directory.
142 if not (all_files - remaining):
143 remaining.difference_update(all_files)
144 wildcards.add(root + os.sep)
146 return set(map(case_map.__getitem__, remaining)) | wildcards
149def compress_for_output_listing(paths: Iterable[str]) -> tuple[set[str], set[str]]:
150 """Returns a tuple of 2 sets of which paths to display to user
152 The first set contains paths that would be deleted. Files of a package
153 are not added and the top-level directory of the package has a '*' added
154 at the end - to signify that all it's contents are removed.
156 The second set contains files that would have been skipped in the above
157 folders.
158 """
160 will_remove = set(paths)
161 will_skip = set()
163 # Determine folders and files
164 folders = set()
165 files = set()
166 for path in will_remove:
167 if path.endswith(".pyc"):
168 continue
169 if path.endswith("__init__.py") or ".dist-info" in path:
170 folders.add(os.path.dirname(path))
171 files.add(path)
173 _normcased_files = set(map(os.path.normcase, files))
175 folders = compact(folders)
177 # This walks the tree using os.walk to not miss extra folders
178 # that might get added.
179 for folder in folders:
180 for dirpath, _, dirfiles in os.walk(folder):
181 for fname in dirfiles:
182 if fname.endswith(".pyc"):
183 continue
185 file_ = os.path.join(dirpath, fname)
186 if (
187 os.path.isfile(file_)
188 and os.path.normcase(file_) not in _normcased_files
189 ):
190 # We are skipping this file. Add it to the set.
191 will_skip.add(file_)
193 will_remove = files | {os.path.join(folder, "*") for folder in folders}
195 return will_remove, will_skip
198class StashedUninstallPathSet:
199 """A set of file rename operations to stash files while
200 tentatively uninstalling them."""
202 def __init__(self) -> None:
203 # Mapping from source file root to [Adjacent]TempDirectory
204 # for files under that directory.
205 self._save_dirs: dict[str, TempDirectory] = {}
206 # (old path, new path) tuples for each move that may need
207 # to be undone.
208 self._moves: list[tuple[str, str]] = []
210 def _get_directory_stash(self, path: str) -> str:
211 """Stashes a directory.
213 Directories are stashed adjacent to their original location if
214 possible, or else moved/copied into the user's temp dir."""
216 try:
217 save_dir: TempDirectory = AdjacentTempDirectory(path)
218 except OSError:
219 save_dir = TempDirectory(kind="uninstall")
220 self._save_dirs[os.path.normcase(path)] = save_dir
222 return save_dir.path
224 def _get_file_stash(self, path: str) -> str:
225 """Stashes a file.
227 If no root has been provided, one will be created for the directory
228 in the user's temp directory."""
229 path = os.path.normcase(path)
230 head, old_head = os.path.dirname(path), None
231 save_dir = None
233 while head != old_head:
234 try:
235 save_dir = self._save_dirs[head]
236 break
237 except KeyError:
238 pass
239 head, old_head = os.path.dirname(head), head
240 else:
241 # Did not find any suitable root
242 head = os.path.dirname(path)
243 save_dir = TempDirectory(kind="uninstall")
244 self._save_dirs[head] = save_dir
246 relpath = os.path.relpath(path, head)
247 if relpath and relpath != os.path.curdir:
248 return os.path.join(save_dir.path, relpath)
249 return save_dir.path
251 def stash(self, path: str) -> str:
252 """Stashes the directory or file and returns its new location.
253 Handle symlinks as files to avoid modifying the symlink targets.
254 """
255 path_is_dir = os.path.isdir(path) and not os.path.islink(path)
256 if path_is_dir:
257 new_path = self._get_directory_stash(path)
258 else:
259 new_path = self._get_file_stash(path)
261 self._moves.append((path, new_path))
262 if path_is_dir and os.path.isdir(new_path):
263 # If we're moving a directory, we need to
264 # remove the destination first or else it will be
265 # moved to inside the existing directory.
266 # We just created new_path ourselves, so it will
267 # be removable.
268 os.rmdir(new_path)
269 renames(path, new_path)
270 return new_path
272 def commit(self) -> None:
273 """Commits the uninstall by removing stashed files."""
274 for save_dir in self._save_dirs.values():
275 save_dir.cleanup()
276 self._moves = []
277 self._save_dirs = {}
279 def rollback(self) -> None:
280 """Undoes the uninstall by moving stashed files back."""
281 for p in self._moves:
282 logger.info("Moving to %s\n from %s", *p)
284 for new_path, path in self._moves:
285 try:
286 logger.debug("Replacing %s from %s", new_path, path)
287 if os.path.isfile(new_path) or os.path.islink(new_path):
288 os.unlink(new_path)
289 elif os.path.isdir(new_path):
290 rmtree(new_path)
291 renames(path, new_path)
292 except OSError as ex:
293 logger.error("Failed to restore %s", new_path)
294 logger.debug("Exception: %s", ex)
296 self.commit()
298 @property
299 def can_rollback(self) -> bool:
300 return bool(self._moves)
303class UninstallPathSet:
304 """A set of file paths to be removed in the uninstallation of a
305 requirement."""
307 def __init__(self, dist: BaseDistribution) -> None:
308 self._paths: set[str] = set()
309 self._refuse: set[str] = set()
310 self._pth: dict[str, UninstallPthEntries] = {}
311 self._dist = dist
312 self._moved_paths = StashedUninstallPathSet()
313 # Create local cache of normalize_path results. Creating an UninstallPathSet
314 # can result in hundreds/thousands of redundant calls to normalize_path with
315 # the same args, which hurts performance.
316 self._normalize_path_cached = functools.lru_cache(normalize_path)
318 def _permitted(self, path: str) -> bool:
319 """
320 Return True if the given path is one we are permitted to
321 remove/modify, False otherwise.
323 """
324 # aka is_local, but caching normalized sys.prefix
325 if not running_under_virtualenv():
326 return True
327 return path.startswith(self._normalize_path_cached(sys.prefix))
329 def add(self, path: str) -> None:
330 head, tail = os.path.split(path)
332 # we normalize the head to resolve parent directory symlinks, but not
333 # the tail, since we only want to uninstall symlinks, not their targets
334 path = os.path.join(self._normalize_path_cached(head), os.path.normcase(tail))
336 if not os.path.exists(path):
337 return
338 if self._permitted(path):
339 self._paths.add(path)
340 else:
341 self._refuse.add(path)
343 # __pycache__ files can show up after 'installed-files.txt' is created,
344 # due to imports
345 if os.path.splitext(path)[1] == ".py":
346 self.add(cache_from_source(path))
348 def add_pth(self, pth_file: str, entry: str) -> None:
349 pth_file = self._normalize_path_cached(pth_file)
350 if self._permitted(pth_file):
351 if pth_file not in self._pth:
352 self._pth[pth_file] = UninstallPthEntries(pth_file)
353 self._pth[pth_file].add(entry)
354 else:
355 self._refuse.add(pth_file)
357 def remove(self, auto_confirm: bool = False, verbose: bool = False) -> None:
358 """Remove paths in ``self._paths`` with confirmation (unless
359 ``auto_confirm`` is True)."""
361 if not self._paths:
362 logger.info(
363 "Can't uninstall '%s'. No files were found to uninstall.",
364 self._dist.raw_name,
365 )
366 return
368 dist_name_version = f"{self._dist.raw_name}-{self._dist.raw_version}"
369 logger.info("Uninstalling %s:", dist_name_version)
371 with indent_log():
372 if auto_confirm or self._allowed_to_proceed(verbose):
373 moved = self._moved_paths
375 for_rename = compress_for_rename(self._paths)
377 for path in sorted(compact(for_rename)):
378 moved.stash(path)
379 logger.verbose("Removing file or directory %s", path)
381 for pth in self._pth.values():
382 pth.remove()
384 logger.info("Successfully uninstalled %s", dist_name_version)
386 def _allowed_to_proceed(self, verbose: bool) -> bool:
387 """Display which files would be deleted and prompt for confirmation"""
389 def _display(msg: str, paths: Iterable[str]) -> None:
390 if not paths:
391 return
393 logger.info(msg)
394 with indent_log():
395 for path in sorted(compact(paths)):
396 logger.info(path)
398 if not verbose:
399 will_remove, will_skip = compress_for_output_listing(self._paths)
400 else:
401 # In verbose mode, display all the files that are going to be
402 # deleted.
403 will_remove = set(self._paths)
404 will_skip = set()
406 _display("Would remove:", will_remove)
407 _display("Would not remove (might be manually added):", will_skip)
408 _display("Would not remove (outside of prefix):", self._refuse)
409 if verbose:
410 _display("Will actually move:", compress_for_rename(self._paths))
412 return ask("Proceed (Y/n)? ", ("y", "n", "")) != "n"
414 def rollback(self) -> None:
415 """Rollback the changes previously made by remove()."""
416 if not self._moved_paths.can_rollback:
417 logger.error(
418 "Can't roll back %s; was not uninstalled",
419 self._dist.raw_name,
420 )
421 return
422 logger.info("Rolling back uninstall of %s", self._dist.raw_name)
423 self._moved_paths.rollback()
424 for pth in self._pth.values():
425 pth.rollback()
427 def commit(self) -> None:
428 """Remove temporary save dir: rollback will no longer be possible."""
429 self._moved_paths.commit()
431 @classmethod
432 def from_dist(cls, dist: BaseDistribution) -> UninstallPathSet:
433 dist_location = dist.location
434 info_location = dist.info_location
435 if dist_location is None:
436 logger.info(
437 "Not uninstalling %s since it is not installed",
438 dist.canonical_name,
439 )
440 return cls(dist)
442 normalized_dist_location = normalize_path(dist_location)
443 if not dist.local:
444 logger.info(
445 "Not uninstalling %s at %s, outside environment %s",
446 dist.canonical_name,
447 normalized_dist_location,
448 sys.prefix,
449 )
450 return cls(dist)
452 if normalized_dist_location in {
453 p
454 for p in {sysconfig.get_path("stdlib"), sysconfig.get_path("platstdlib")}
455 if p
456 }:
457 logger.info(
458 "Not uninstalling %s at %s, as it is in the standard library.",
459 dist.canonical_name,
460 normalized_dist_location,
461 )
462 return cls(dist)
464 paths_to_remove = cls(dist)
465 develop_egg_link = egg_link_path_from_location(dist.raw_name)
467 # Distribution is installed with metadata in a "flat" .egg-info
468 # directory. This means it is not a modern .dist-info installation, an
469 # egg, or legacy editable.
470 setuptools_flat_installation = (
471 dist.installed_with_setuptools_egg_info
472 and info_location is not None
473 and os.path.exists(info_location)
474 # If dist is editable and the location points to a ``.egg-info``,
475 # we are in fact in the legacy editable case.
476 and not info_location.endswith(f"{dist.setuptools_filename}.egg-info")
477 )
479 # Uninstall cases order do matter as in the case of 2 installs of the
480 # same package, pip needs to uninstall the currently detected version
481 if setuptools_flat_installation:
482 if info_location is not None:
483 paths_to_remove.add(info_location)
484 installed_files = dist.iter_declared_entries()
485 if installed_files is not None:
486 for installed_file in installed_files:
487 paths_to_remove.add(os.path.join(dist_location, installed_file))
488 # FIXME: need a test for this elif block
489 # occurs with --single-version-externally-managed/--record outside
490 # of pip
491 elif dist.is_file("top_level.txt"):
492 try:
493 namespace_packages = dist.read_text("namespace_packages.txt")
494 except FileNotFoundError:
495 namespaces = []
496 else:
497 namespaces = namespace_packages.splitlines(keepends=False)
498 for top_level_pkg in [
499 p
500 for p in dist.read_text("top_level.txt").splitlines()
501 if p and p not in namespaces
502 ]:
503 path = os.path.join(dist_location, top_level_pkg)
504 paths_to_remove.add(path)
505 paths_to_remove.add(f"{path}.py")
506 paths_to_remove.add(f"{path}.pyc")
507 paths_to_remove.add(f"{path}.pyo")
509 elif dist.installed_by_distutils:
510 raise LegacyDistutilsInstall(distribution=dist)
512 elif dist.installed_as_egg:
513 # package installed by easy_install
514 # We cannot match on dist.egg_name because it can slightly vary
515 # i.e. setuptools-0.6c11-py2.6.egg vs setuptools-0.6rc11-py2.6.egg
516 # XXX We use normalized_dist_location because dist_location my contain
517 # a trailing / if the distribution is a zipped egg
518 # (which is not a directory).
519 paths_to_remove.add(normalized_dist_location)
520 easy_install_egg = os.path.split(normalized_dist_location)[1]
521 easy_install_pth = os.path.join(
522 os.path.dirname(normalized_dist_location),
523 "easy-install.pth",
524 )
525 paths_to_remove.add_pth(easy_install_pth, "./" + easy_install_egg)
527 elif dist.installed_with_dist_info:
528 for path in uninstallation_paths(dist):
529 paths_to_remove.add(path)
531 elif develop_egg_link:
532 # PEP 660 modern editable is handled in the ``.dist-info`` case
533 # above, so this only covers the setuptools-style editable.
534 with open(develop_egg_link) as fh:
535 link_pointer = os.path.normcase(fh.readline().strip())
536 normalized_link_pointer = paths_to_remove._normalize_path_cached(
537 link_pointer
538 )
539 assert os.path.samefile(
540 normalized_link_pointer, normalized_dist_location
541 ), (
542 f"Egg-link {develop_egg_link} (to {link_pointer}) does not match "
543 f"installed location of {dist.raw_name} (at {dist_location})"
544 )
545 paths_to_remove.add(develop_egg_link)
546 easy_install_pth = os.path.join(
547 os.path.dirname(develop_egg_link), "easy-install.pth"
548 )
549 paths_to_remove.add_pth(easy_install_pth, dist_location)
551 else:
552 logger.debug(
553 "Not sure how to uninstall: %s - Check: %s",
554 dist,
555 dist_location,
556 )
558 if dist.in_usersite:
559 bin_dir = get_bin_user()
560 else:
561 bin_dir = get_bin_prefix()
563 # find distutils scripts= scripts
564 try:
565 for script in dist.iter_distutils_script_names():
566 paths_to_remove.add(os.path.join(bin_dir, script))
567 if WINDOWS:
568 paths_to_remove.add(os.path.join(bin_dir, f"{script}.bat"))
569 except (FileNotFoundError, NotADirectoryError):
570 pass
572 # find console_scripts and gui_scripts
573 def iter_scripts_to_remove(
574 dist: BaseDistribution,
575 bin_dir: str,
576 ) -> Generator[str, None, None]:
577 for entry_point in dist.iter_entry_points():
578 if entry_point.group == "console_scripts":
579 yield from _script_names(bin_dir, entry_point.name, False)
580 elif entry_point.group == "gui_scripts":
581 yield from _script_names(bin_dir, entry_point.name, True)
583 for s in iter_scripts_to_remove(dist, bin_dir):
584 paths_to_remove.add(s)
586 return paths_to_remove
589class UninstallPthEntries:
590 def __init__(self, pth_file: str) -> None:
591 self.file = pth_file
592 self.entries: set[str] = set()
593 self._saved_lines: list[bytes] | None = None
595 def add(self, entry: str) -> None:
596 entry = os.path.normcase(entry)
597 # On Windows, os.path.normcase converts the entry to use
598 # backslashes. This is correct for entries that describe absolute
599 # paths outside of site-packages, but all the others use forward
600 # slashes.
601 # os.path.splitdrive is used instead of os.path.isabs because isabs
602 # treats non-absolute paths with drive letter markings like c:foo\bar
603 # as absolute paths. It also does not recognize UNC paths if they don't
604 # have more than "\\sever\share". Valid examples: "\\server\share\" or
605 # "\\server\share\folder".
606 if WINDOWS and not os.path.splitdrive(entry)[0]:
607 entry = entry.replace("\\", "/")
608 self.entries.add(entry)
610 def remove(self) -> None:
611 logger.verbose("Removing pth entries from %s:", self.file)
613 # If the file doesn't exist, log a warning and return
614 if not os.path.isfile(self.file):
615 logger.warning("Cannot remove entries from nonexistent file %s", self.file)
616 return
617 with open(self.file, "rb") as fh:
618 # windows uses '\r\n' with py3k, but uses '\n' with py2.x
619 lines = fh.readlines()
620 self._saved_lines = lines
621 if any(b"\r\n" in line for line in lines):
622 endline = "\r\n"
623 else:
624 endline = "\n"
625 # handle missing trailing newline
626 if lines and not lines[-1].endswith(endline.encode("utf-8")):
627 lines[-1] = lines[-1] + endline.encode("utf-8")
628 for entry in self.entries:
629 try:
630 logger.verbose("Removing entry: %s", entry)
631 lines.remove((entry + endline).encode("utf-8"))
632 except ValueError:
633 pass
634 with open(self.file, "wb") as fh:
635 fh.writelines(lines)
637 def rollback(self) -> bool:
638 if self._saved_lines is None:
639 logger.error("Cannot roll back changes to %s, none were made", self.file)
640 return False
641 logger.debug("Rolling %s back to previous state", self.file)
642 with open(self.file, "wb") as fh:
643 fh.writelines(self._saved_lines)
644 return True