Source code for pip._internal.req.req_tracker

from __future__ import absolute_import

import contextlib
import errno
import hashlib
import logging
import os

from pip._vendor import contextlib2

from pip._internal.utils.temp_dir import TempDirectory
from pip._internal.utils.typing import MYPY_CHECK_RUNNING

    from types import TracebackType
    from typing import Dict, Iterator, Optional, Set, Type, Union
    from pip._internal.req.req_install import InstallRequirement
    from import Link

logger = logging.getLogger(__name__)

def update_env_context_manager(**changes):
    # type: (str) -> Iterator[None]
    target = os.environ

    # Save values from the target and change them.
    non_existent_marker = object()
    saved_values = {}  # type: Dict[str, Union[object, str]]
    for name, new_value in changes.items():
            saved_values[name] = target[name]
        except KeyError:
            saved_values[name] = non_existent_marker
        target[name] = new_value

        # Restore original values in the target.
        for name, original_value in saved_values.items():
            if original_value is non_existent_marker:
                del target[name]
                assert isinstance(original_value, str)  # for mypy
                target[name] = original_value

[docs]@contextlib.contextmanager def get_requirement_tracker(): # type: () -> Iterator[RequirementTracker] root = os.environ.get('PIP_REQ_TRACKER') with contextlib2.ExitStack() as ctx: if root is None: root = ctx.enter_context( TempDirectory(kind='req-tracker') ).path ctx.enter_context(update_env_context_manager(PIP_REQ_TRACKER=root)) logger.debug("Initialized build tracking at %s", root) with RequirementTracker(root) as tracker: yield tracker
[docs]class RequirementTracker(object): def __init__(self, root): # type: (str) -> None self._root = root self._entries = set() # type: Set[InstallRequirement] logger.debug("Created build tracker: %s", self._root) def __enter__(self): # type: () -> RequirementTracker logger.debug("Entered build tracker: %s", self._root) return self def __exit__( self, exc_type, # type: Optional[Type[BaseException]] exc_val, # type: Optional[BaseException] exc_tb # type: Optional[TracebackType] ): # type: (...) -> None self.cleanup()
[docs] def _entry_path(self, link): # type: (Link) -> str hashed = hashlib.sha224(link.url_without_fragment.encode()).hexdigest() return os.path.join(self._root, hashed)
[docs] def add(self, req): # type: (InstallRequirement) -> None """Add an InstallRequirement to build tracking. """ assert # Get the file to write information about this requirement. entry_path = self._entry_path( # Try reading from the file. If it exists and can be read from, a build # is already in progress, so a LookupError is raised. try: with open(entry_path) as fp: contents = except IOError as e: # if the error is anything other than "file does not exist", raise. if e.errno != errno.ENOENT: raise else: message = '{} is already being built: {}'.format(, contents) raise LookupError(message) # If we're here, req should really not be building already. assert req not in self._entries # Start tracking this requirement. with open(entry_path, 'w') as fp: fp.write(str(req)) self._entries.add(req) logger.debug('Added %s to build tracker %r', req, self._root)
[docs] def remove(self, req): # type: (InstallRequirement) -> None """Remove an InstallRequirement from build tracking. """ assert # Delete the created file and the corresponding entries. os.unlink(self._entry_path( self._entries.remove(req) logger.debug('Removed %s from build tracker %r', req, self._root)
[docs] def cleanup(self): # type: () -> None for req in set(self._entries): self.remove(req) logger.debug("Removed build tracker: %r", self._root)
[docs] @contextlib.contextmanager def track(self, req): # type: (InstallRequirement) -> Iterator[None] self.add(req) yield self.remove(req)