# The following comment should be removed at some point in the future.
# mypy: strict-optional=False

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

if MYPY_CHECK_RUNNING:
    from types import TracebackType
    from typing import Iterator, Optional, Set, Type
    from pip._internal.req.req_install import InstallRequirement
    from pip._internal.models.link import Link

logger = logging.getLogger(__name__)


@contextlib.contextmanager
def update_env_context_manager(**changes):
    target = os.environ

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

    try:
        yield
    finally:
        # Restore original values in the target.
        for name, value in saved_values.items():
            if value is non_existent_marker:
                del target[name]
            else:
                target[name] = value


@contextlib.contextmanager
def get_requirement_tracker():
    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))

        yield RequirementTracker(root)


class RequirementTracker(object):

    def __init__(self):
        # type: () -> None
        self._root = os.environ.get('PIP_REQ_TRACKER')
        if self._root is None:
            self._temp_dir = TempDirectory(delete=False, kind='req-tracker')
            self._root = os.environ['PIP_REQ_TRACKER'] = self._temp_dir.path
            logger.debug('Created requirements tracker %r', self._root)
        else:
            self._temp_dir = None
            logger.debug('Re-using requirements tracker %r', self._root)
        self._entries = set()  # type: Set[InstallRequirement]

    def __enter__(self):
        # type: () -> RequirementTracker
        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()

    def _entry_path(self, link):
        # type: (Link) -> str
        hashed = hashlib.sha224(link.url_without_fragment.encode()).hexdigest()
        return os.path.join(self._root, hashed)

    def add(self, req):
        # type: (InstallRequirement) -> None
        """Add an InstallRequirement to build tracking.
        """

        # Get the file to write information about this requirement.
        entry_path = self._entry_path(req.link)

        # 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 = fp.read()
        except IOError as e:
            # if the error is anything other than "file does not exist", raise.
            if e.errno != errno.ENOENT:
                raise
        else:
            message = '%s is already being built: %s' % (req.link, 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)

    def remove(self, req):
        # type: (InstallRequirement) -> None
        """Remove an InstallRequirement from build tracking.
        """

        # Delete the created file and the corresponding entries.
        os.unlink(self._entry_path(req.link))
        self._entries.remove(req)

        logger.debug('Removed %s from build tracker %r', req, self._root)

    def cleanup(self):
        # type: () -> None
        for req in set(self._entries):
            self.remove(req)

        if self._temp_dir is None:
            # Did not setup the directory. No action needed.
            logger.debug("Cleaned build tracker: %r", self._root)
            return

        # Cleanup the directory.
        self._temp_dir.cleanup()
        logger.debug("Removed build tracker: %r", self._root)

    @contextlib.contextmanager
    def track(self, req):
        # type: (InstallRequirement) -> Iterator[None]
        self.add(req)
        yield
        self.remove(req)

@@ -1751,24 +1751,30 @@
 _init__(self
+, root
 ):%0A        #
@@ -1785,450 +1785,46 @@
 e: (
-) -%3E None%0A        self._root = os.environ.get('PIP_REQ_TRACKER')%0A        if self._root is None:%0A            self._temp_dir = TempDirectory(delete=False, kind='req-tracker')%0A            self._root = os.environ%5B'PIP_REQ_TRACKER'%5D = self._temp_dir.path%0A            logger.debug('Created requirements tracker %25r', self._root)%0A        else:%0A            self._temp_dir = None%0A            logger.debug('Re-using requirements tracker %25r', self._root)
+str) -%3E None%0A        self._root = root
 %0A   
@@ -4019,256 +4019,8 @@
 q)%0A%0A
-        if self._temp_dir is None:%0A            # Did not setup the directory. No action needed.%0A            logger.debug(%22Cleaned build tracker: %25r%22, self._root)%0A            return%0A%0A        # Cleanup the directory.%0A        self._temp_dir.cleanup()%0A
     

Drop environment and directory management from RequirementTracker