Add VCS Support

fetchcode/vcs/__init__.py

@@ -0,0 +1,63 @@
+# fetchcode is a free software tool from nexB Inc. and others.
+# Visit https://github.com/nexB/fetchcode for support and download.
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# http://nexb.com and http://aboutcode.org
+#
+# This software is licensed under the Apache License version 2.0.
+#
+# You may not use this software except in compliance with the License.
+# You may obtain a copy of the License at:
+# http://apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software distributed
+# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+# CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+
+import os
+import tempfile
+from urllib.parse import urlparse
+
+from fetchcode.vcs.pip._internal.vcs.bazaar import Bazaar
+from fetchcode.vcs.pip._internal.vcs.git import Git
+from fetchcode.vcs.pip._internal.vcs.mercurial import Mercurial
+from fetchcode.vcs.pip._internal.utils import misc
+from fetchcode.vcs.pip._internal.vcs.subversion import Subversion
+from fetchcode.vcs.pip._internal.vcs import vcs
+
+
+class VCSResponse:
+    """
+    Represent the response from fetching a VCS URL with:
+- `dest_dir`: destination of directory
+- `vcs_type`: VCS Type of URL (git,bzr,hg,svn)
+- `domain` : Source of git VCS (GitHub, Gitlab, Bitbucket)
+    """
+
+    def __init__(self, dest_dir, vcs_type, domain):
+        self.dest_dir = dest_dir
+        self.vcs_type = vcs_type
+        self.domain = domain
+
+
+def fetch_via_vcs(url):
+    """
+    Take `url` as input and store the content of it at location specified at `location` string
+    Return a VCSResponse object 
+    """
+    parsed_url = urlparse(url)
+    scheme = parsed_url.scheme
+    domain = parsed_url.netloc
+    temp = tempfile.mkdtemp()
+    os.rmdir(temp)
+    if scheme not in vcs.all_schemes:
+        raise Exception("Not a supported/known scheme.")
+
+    for vcs_name, vcs_backend in vcs._registry.items():
+        if scheme in vcs_backend.schemes:
+            vcs_type = vcs_name
+
+    backend = vcs.get_backend_for_scheme(scheme)
+    backend.obtain(dest=temp, url=misc.hide_url(url))
+
+    return VCSResponse(dest_dir=temp, vcs_type=vcs_type, domain=domain)

fetchcode/vcs/pip/__init__.py

@@ -0,0 +1,18 @@
+from fetchcode.vcs.pip._internal.utils.typing import MYPY_CHECK_RUNNING
+
+if MYPY_CHECK_RUNNING:
+    from typing import List, Optional
+
+
+__version__ = "20.1.1"
+
+
+def main(args=None):
+    # type: (Optional[List[str]]) -> int
+    """This is an internal API only meant for use by pip's own console scripts.
+
+    For additional details, see https://github.com/pypa/pip/issues/7498.
+    """
+    from fetchcode.vcs.pip._internal.utils.entrypoints import _wrapper
+
+    return _wrapper(args)

fetchcode/vcs/pip/__main__.py

@@ -0,0 +1,26 @@
+from __future__ import absolute_import
+
+import os
+import sys
+
+# Remove '' and current working directory from the first entry
+# of sys.path, if present to avoid using current directory
+# in pip commands check, freeze, install, list and show,
+# when invoked as python -m pip <command>
+if sys.path[0] in ('', os.getcwd()):
+    sys.path.pop(0)
+
+# If we are running from a wheel, add the wheel to sys.path
+# This allows the usage python pip-*.whl/pip install pip-*.whl
+if __package__ == '':
+    # __file__ is pip-*.whl/pip/__main__.py
+    # first dirname call strips of '/__main__.py', second strips off '/pip'
+    # Resulting path is the name of the wheel itself
+    # Add that to sys.path so we can import pip
+    path = os.path.dirname(os.path.dirname(__file__))
+    sys.path.insert(0, path)
+
+from fetchcode.vcs.pip._internal.cli.main import main as _main  # isort:skip # noqa
+
+if __name__ == '__main__':
+    sys.exit(_main())

fetchcode/vcs/pip/_internal/__init__.py

@@ -0,0 +1,17 @@
+import fetchcode.vcs.pip._internal.utils.inject_securetransport  # noqa
+from fetchcode.vcs.pip._internal.utils.typing import MYPY_CHECK_RUNNING
+
+if MYPY_CHECK_RUNNING:
+    from typing import Optional, List
+
+
+def main(args=None):
+    # type: (Optional[List[str]]) -> int
+    """This is preserved for old console scripts that may still be referencing
+    it.
+
+    For additional details, see https://github.com/pypa/pip/issues/7498.
+    """
+    from fetchcode.vcs.pip._internal.utils.entrypoints import _wrapper
+
+    return _wrapper(args)

fetchcode/vcs/pip/_internal/build_env.py

@@ -0,0 +1,219 @@
+"""Build Environment used for isolation during sdist building
+"""
+
+# The following comment should be removed at some point in the future.
+# mypy: strict-optional=False
+# mypy: disallow-untyped-defs=False
+
+import logging
+import os
+import sys
+import textwrap
+from collections import OrderedDict
+from distutils.sysconfig import get_python_lib
+from sysconfig import get_paths
+
+from fetchcode.vcs.pip._vendor.pkg_resources import Requirement, VersionConflict, WorkingSet
+
+from pip import __file__ as pip_location
+from fetchcode.vcs.pip._internal.cli.spinners import open_spinner
+from fetchcode.vcs.pip._internal.utils.subprocess import call_subprocess
+from fetchcode.vcs.pip._internal.utils.temp_dir import TempDirectory, tempdir_kinds
+from fetchcode.vcs.pip._internal.utils.typing import MYPY_CHECK_RUNNING
+
+if MYPY_CHECK_RUNNING:
+    from typing import Tuple, Set, Iterable, Optional, List
+    from fetchcode.vcs.pip._internal.index.package_finder import PackageFinder
+
+logger = logging.getLogger(__name__)
+
+
+class _Prefix:
+
+    def __init__(self, path):
+        # type: (str) -> None
+        self.path = path
+        self.setup = False
+        self.bin_dir = get_paths(
+            'nt' if os.name == 'nt' else 'posix_prefix',
+            vars={'base': path, 'platbase': path}
+        )['scripts']
+        # Note: prefer distutils' sysconfig to get the
+        # library paths so PyPy is correctly supported.
+        purelib = get_python_lib(plat_specific=False, prefix=path)
+        platlib = get_python_lib(plat_specific=True, prefix=path)
+        if purelib == platlib:
+            self.lib_dirs = [purelib]
+        else:
+            self.lib_dirs = [purelib, platlib]
+
+
+class BuildEnvironment(object):
+    """Creates and manages an isolated environment to install build deps
+    """
+
+    def __init__(self):
+        # type: () -> None
+        temp_dir = TempDirectory(
+            kind=tempdir_kinds.BUILD_ENV, globally_managed=True
+        )
+
+        self._prefixes = OrderedDict((
+            (name, _Prefix(os.path.join(temp_dir.path, name)))
+            for name in ('normal', 'overlay')
+        ))
+
+        self._bin_dirs = []  # type: List[str]
+        self._lib_dirs = []  # type: List[str]
+        for prefix in reversed(list(self._prefixes.values())):
+            self._bin_dirs.append(prefix.bin_dir)
+            self._lib_dirs.extend(prefix.lib_dirs)
+
+        # Customize site to:
+        # - ensure .pth files are honored
+        # - prevent access to system site packages
+        system_sites = {
+            os.path.normcase(site) for site in (
+                get_python_lib(plat_specific=False),
+                get_python_lib(plat_specific=True),
+            )
+        }
+        self._site_dir = os.path.join(temp_dir.path, 'site')
+        if not os.path.exists(self._site_dir):
+            os.mkdir(self._site_dir)
+        with open(os.path.join(self._site_dir, 'sitecustomize.py'), 'w') as fp:
+            fp.write(textwrap.dedent(
+                '''
+                import os, site, sys
+
+                # First, drop system-sites related paths.
+                original_sys_path = sys.path[:]
+                known_paths = set()
+                for path in {system_sites!r}:
+                    site.addsitedir(path, known_paths=known_paths)
+                system_paths = set(
+                    os.path.normcase(path)
+                    for path in sys.path[len(original_sys_path):]
+                )
+                original_sys_path = [
+                    path for path in original_sys_path
+                    if os.path.normcase(path) not in system_paths
+                ]
+                sys.path = original_sys_path
+
+                # Second, add lib directories.
+                # ensuring .pth file are processed.
+                for path in {lib_dirs!r}:
+                    assert not path in sys.path
+                    site.addsitedir(path)
+                '''
+            ).format(system_sites=system_sites, lib_dirs=self._lib_dirs))
+
+    def __enter__(self):
+        self._save_env = {
+            name: os.environ.get(name, None)
+            for name in ('PATH', 'PYTHONNOUSERSITE', 'PYTHONPATH')
+        }
+
+        path = self._bin_dirs[:]
+        old_path = self._save_env['PATH']
+        if old_path:
+            path.extend(old_path.split(os.pathsep))
+
+        pythonpath = [self._site_dir]
+
+        os.environ.update({
+            'PATH': os.pathsep.join(path),
+            'PYTHONNOUSERSITE': '1',
+            'PYTHONPATH': os.pathsep.join(pythonpath),
+        })
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        for varname, old_value in self._save_env.items():
+            if old_value is None:
+                os.environ.pop(varname, None)
+            else:
+                os.environ[varname] = old_value
+
+    def check_requirements(self, reqs):
+        # type: (Iterable[str]) -> Tuple[Set[Tuple[str, str]], Set[str]]
+        """Return 2 sets:
+            - conflicting requirements: set of (installed, wanted) reqs tuples
+            - missing requirements: set of reqs
+        """
+        missing = set()
+        conflicting = set()
+        if reqs:
+            ws = WorkingSet(self._lib_dirs)
+            for req in reqs:
+                try:
+                    if ws.find(Requirement.parse(req)) is None:
+                        missing.add(req)
+                except VersionConflict as e:
+                    conflicting.add((str(e.args[0].as_requirement()),
+                                     str(e.args[1])))
+        return conflicting, missing
+
+    def install_requirements(
+        self,
+        finder,  # type: PackageFinder
+        requirements,  # type: Iterable[str]
+        prefix_as_string,  # type: str
+        message  # type: Optional[str]
+    ):
+        # type: (...) -> None
+        prefix = self._prefixes[prefix_as_string]
+        assert not prefix.setup
+        prefix.setup = True
+        if not requirements:
+            return
+        args = [
+            sys.executable, os.path.dirname(pip_location), 'install',
+            '--ignore-installed', '--no-user', '--prefix', prefix.path,
+            '--no-warn-script-location',
+        ]  # type: List[str]
+        if logger.getEffectiveLevel() <= logging.DEBUG:
+            args.append('-v')
+        for format_control in ('no_binary', 'only_binary'):
+            formats = getattr(finder.format_control, format_control)
+            args.extend(('--' + format_control.replace('_', '-'),
+                         ','.join(sorted(formats or {':none:'}))))
+
+        index_urls = finder.index_urls
+        if index_urls:
+            args.extend(['-i', index_urls[0]])
+            for extra_index in index_urls[1:]:
+                args.extend(['--extra-index-url', extra_index])
+        else:
+            args.append('--no-index')
+        for link in finder.find_links:
+            args.extend(['--find-links', link])
+
+        for host in finder.trusted_hosts:
+            args.extend(['--trusted-host', host])
+        if finder.allow_all_prereleases:
+            args.append('--pre')
+        args.append('--')
+        args.extend(requirements)
+        with open_spinner(message) as spinner:
+            call_subprocess(args, spinner=spinner)
+
+
+class NoOpBuildEnvironment(BuildEnvironment):
+    """A no-op drop-in replacement for BuildEnvironment
+    """
+
+    def __init__(self):
+        pass
+
+    def __enter__(self):
+        pass
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        pass
+
+    def cleanup(self):
+        pass
+
+    def install_requirements(self, finder, requirements, prefix, message):
+        raise NotImplementedError()