Mercurial > repos > guerler > hhblits
diff lib/python3.8/site-packages/pip/_internal/utils/temp_dir.py @ 0:9e54283cc701 draft
"planemo upload commit d12c32a45bcd441307e632fca6d9af7d60289d44"
author | guerler |
---|---|
date | Mon, 27 Jul 2020 03:47:31 -0400 |
parents | |
children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/lib/python3.8/site-packages/pip/_internal/utils/temp_dir.py Mon Jul 27 03:47:31 2020 -0400 @@ -0,0 +1,250 @@ +from __future__ import absolute_import + +import errno +import itertools +import logging +import os.path +import tempfile +from contextlib import contextmanager + +from pip._vendor.contextlib2 import ExitStack + +from pip._internal.utils.misc import rmtree +from pip._internal.utils.typing import MYPY_CHECK_RUNNING + +if MYPY_CHECK_RUNNING: + from typing import Any, Dict, Iterator, Optional, TypeVar + + _T = TypeVar('_T', bound='TempDirectory') + + +logger = logging.getLogger(__name__) + + +_tempdir_manager = None # type: Optional[ExitStack] + + +@contextmanager +def global_tempdir_manager(): + # type: () -> Iterator[None] + global _tempdir_manager + with ExitStack() as stack: + old_tempdir_manager, _tempdir_manager = _tempdir_manager, stack + try: + yield + finally: + _tempdir_manager = old_tempdir_manager + + +class TempDirectoryTypeRegistry(object): + """Manages temp directory behavior + """ + + def __init__(self): + # type: () -> None + self._should_delete = {} # type: Dict[str, bool] + + def set_delete(self, kind, value): + # type: (str, bool) -> None + """Indicate whether a TempDirectory of the given kind should be + auto-deleted. + """ + self._should_delete[kind] = value + + def get_delete(self, kind): + # type: (str) -> bool + """Get configured auto-delete flag for a given TempDirectory type, + default True. + """ + return self._should_delete.get(kind, True) + + +_tempdir_registry = None # type: Optional[TempDirectoryTypeRegistry] + + +@contextmanager +def tempdir_registry(): + # type: () -> Iterator[TempDirectoryTypeRegistry] + """Provides a scoped global tempdir registry that can be used to dictate + whether directories should be deleted. + """ + global _tempdir_registry + old_tempdir_registry = _tempdir_registry + _tempdir_registry = TempDirectoryTypeRegistry() + try: + yield _tempdir_registry + finally: + _tempdir_registry = old_tempdir_registry + + +class TempDirectory(object): + """Helper class that owns and cleans up a temporary directory. + + This class can be used as a context manager or as an OO representation of a + temporary directory. + + Attributes: + path + Location to the created temporary directory + delete + Whether the directory should be deleted when exiting + (when used as a contextmanager) + + Methods: + cleanup() + Deletes the temporary directory + + When used as a context manager, if the delete attribute is True, on + exiting the context the temporary directory is deleted. + """ + + def __init__( + self, + path=None, # type: Optional[str] + delete=None, # type: Optional[bool] + kind="temp", # type: str + globally_managed=False, # type: bool + ): + super(TempDirectory, self).__init__() + + # If we were given an explicit directory, resolve delete option now. + # Otherwise we wait until cleanup and see what tempdir_registry says. + if path is not None and delete is None: + delete = False + + if path is None: + path = self._create(kind) + + self._path = path + self._deleted = False + self.delete = delete + self.kind = kind + + if globally_managed: + assert _tempdir_manager is not None + _tempdir_manager.enter_context(self) + + @property + def path(self): + # type: () -> str + assert not self._deleted, ( + "Attempted to access deleted path: {}".format(self._path) + ) + return self._path + + def __repr__(self): + # type: () -> str + return "<{} {!r}>".format(self.__class__.__name__, self.path) + + def __enter__(self): + # type: (_T) -> _T + return self + + def __exit__(self, exc, value, tb): + # type: (Any, Any, Any) -> None + if self.delete is not None: + delete = self.delete + elif _tempdir_registry: + delete = _tempdir_registry.get_delete(self.kind) + else: + delete = True + + if delete: + self.cleanup() + + def _create(self, kind): + # type: (str) -> str + """Create a temporary directory and store its path in self.path + """ + # We realpath here because some systems have their default tmpdir + # symlinked to another directory. This tends to confuse build + # scripts, so we canonicalize the path by traversing potential + # symlinks here. + path = os.path.realpath( + tempfile.mkdtemp(prefix="pip-{}-".format(kind)) + ) + logger.debug("Created temporary directory: {}".format(path)) + return path + + def cleanup(self): + # type: () -> None + """Remove the temporary directory created and reset state + """ + self._deleted = True + if os.path.exists(self._path): + rmtree(self._path) + + +class AdjacentTempDirectory(TempDirectory): + """Helper class that creates a temporary directory adjacent to a real one. + + Attributes: + original + The original directory to create a temp directory for. + path + After calling create() or entering, contains the full + path to the temporary directory. + delete + Whether the directory should be deleted when exiting + (when used as a contextmanager) + + """ + # The characters that may be used to name the temp directory + # We always prepend a ~ and then rotate through these until + # a usable name is found. + # pkg_resources raises a different error for .dist-info folder + # with leading '-' and invalid metadata + LEADING_CHARS = "-~.=%0123456789" + + def __init__(self, original, delete=None): + # type: (str, Optional[bool]) -> None + self.original = original.rstrip('/\\') + super(AdjacentTempDirectory, self).__init__(delete=delete) + + @classmethod + def _generate_names(cls, name): + # type: (str) -> Iterator[str] + """Generates a series of temporary names. + + The algorithm replaces the leading characters in the name + with ones that are valid filesystem characters, but are not + valid package names (for both Python and pip definitions of + package). + """ + for i in range(1, len(name)): + for candidate in itertools.combinations_with_replacement( + cls.LEADING_CHARS, i - 1): + new_name = '~' + ''.join(candidate) + name[i:] + if new_name != name: + yield new_name + + # If we make it this far, we will have to make a longer name + for i in range(len(cls.LEADING_CHARS)): + for candidate in itertools.combinations_with_replacement( + cls.LEADING_CHARS, i): + new_name = '~' + ''.join(candidate) + name + if new_name != name: + yield new_name + + def _create(self, kind): + # type: (str) -> str + root, name = os.path.split(self.original) + for candidate in self._generate_names(name): + path = os.path.join(root, candidate) + try: + os.mkdir(path) + except OSError as ex: + # Continue if the name exists already + if ex.errno != errno.EEXIST: + raise + else: + path = os.path.realpath(path) + break + else: + # Final fallback on the default behavior. + path = os.path.realpath( + tempfile.mkdtemp(prefix="pip-{}-".format(kind)) + ) + + logger.debug("Created temporary directory: {}".format(path)) + return path