Add --python-executable option for pip-sync

piptools/scripts/sync.py

@@ -15,7 +15,7 @@
 from ..exceptions import PipToolsError
 from ..logging import log
 from ..repositories import PyPIRepository
-from ..utils import flat_map
+from ..utils import flat_map, get_sys_path_for_python_executable
 
 DEFAULT_REQUIREMENTS_FILE = "requirements.txt"
 
@@ -55,6 +55,10 @@
     is_flag=True,
     help="Ignore package index (only looking at --find-links URLs instead)",
 )
+@click.option(
+    "--python-executable",
+    help="Custom python executable path if targeting an environment other than current",
+)
 @click.option("-v", "--verbose", count=True, help="Show more output")
 @click.option("-q", "--quiet", count=True, help="Give less output")
 @click.option(
@@ -77,6 +81,7 @@ def cli(
     extra_index_url: Tuple[str, ...],
     trusted_host: Tuple[str, ...],
     no_index: bool,
+    python_executable: Optional[str],
     verbose: int,
     quiet: int,
     user_only: bool,
@@ -108,6 +113,11 @@ def cli(
             log.error("ERROR: " + msg)
             sys.exit(2)
 
+    if python_executable is not None and not os.path.exists(python_executable):
+        msg = "Python executable {} not found"
+        log.error(msg.format(python_executable))
+        sys.exit(2)
+
     install_command = cast(InstallCommand, create_command("install"))
     options, _ = install_command.parse_args([])
     session = install_command._build_session(options)
@@ -125,7 +135,13 @@ def cli(
         log.error(str(e))
         sys.exit(2)
 
-    installed_dists = get_installed_distributions(skip=[], user_only=user_only)
+    paths = None
+    if python_executable:
+        paths = get_sys_path_for_python_executable(python_executable)
+
+    installed_dists = get_installed_distributions(
+        skip=[], user_only=user_only, paths=paths, local_only=python_executable is None
+    )
     to_install, to_uninstall = sync.diff(merged_requirements, installed_dists)
 
     install_flags = (
@@ -149,6 +165,7 @@ def cli(
             dry_run=dry_run,
             install_flags=install_flags,
             ask=ask,
+            python_executable=python_executable,
         )
     )
 

piptools/sync.py

@@ -178,12 +178,15 @@ def sync(
     dry_run: bool = False,
     install_flags: Optional[List[str]] = None,
     ask: bool = False,
+    python_executable: Optional[str] = None,
 ) -> int:
     """
     Install and uninstalls the given sets of modules.
     """
     exit_code = 0
 
+    python_executable = python_executable or sys.executable
+
     if not to_uninstall and not to_install:
         log.info("Everything up-to-date", err=False)
         return exit_code
@@ -216,7 +219,7 @@ def sync(
         if to_uninstall:
             run(  # nosec
                 [
-                    sys.executable,
+                    python_executable,
                     "-m",
                     "pip",
                     "uninstall",
@@ -244,7 +247,7 @@ def sync(
             try:
                 run(  # nosec
                     [
-                        sys.executable,
+                        python_executable,
                         "-m",
                         "pip",
                         "install",

piptools/utils.py

@@ -1,11 +1,15 @@
 import collections
 import itertools
+import json
+import os
 import shlex
+import subprocess  # nosec
 from typing import (
     Callable,
     Dict,
     Iterable,
     Iterator,
+    List,
     Optional,
     Set,
     Tuple,
@@ -306,3 +310,16 @@ def get_compile_command(click_ctx: click.Context) -> str:
                     left_args.append(f"{option_long_name}={shlex.quote(str(val))}")
 
     return " ".join(["pip-compile", *sorted(left_args), *sorted(right_args)])
+
+
+def get_sys_path_for_python_executable(python_executable: str) -> List[str]:
+    """
+    Returns sys.path list for the given python executable.
+    """
+    result = subprocess.check_output(  # nosec
+        [python_executable, "-c", "import sys;import json;print(json.dumps(sys.path))"]
+    )
+    paths = json.loads(result)
+    assert isinstance(paths, list)
+    assert all(isinstance(i, str) for i in paths)
+    return [os.path.abspath(path) for path in paths]

tests/test_cli_sync.py

@@ -242,3 +242,78 @@ def test_sync_dry_run_returns_non_zero_exit_code(runner):
     out = runner.invoke(cli, ["--dry-run"])
 
     assert out.exit_code == 1
+
+
+@mock.patch("piptools.scripts.sync.get_sys_path_for_python_executable")
+@mock.patch("piptools.scripts.sync.get_installed_distributions")
+@mock.patch("piptools.sync.run")
+def test_python_executable_option(
+    run,
+    get_installed_distributions,
+    get_sys_path_for_python_executable,
+    runner,
+    fake_dist,
+):
+    """
+    Make sure sync command can run with `--python-executable` option
+    """
+    with open("requirements.txt", "w") as req_in:
+        req_in.write("small-fake-a==1.10.0")
+
+    custom_executable = "custom_executable"
+    with open(custom_executable, "w") as exec_file:
+        exec_file.write("")
+
+    sys_paths = ["", "./"]
+    get_sys_path_for_python_executable.return_value = sys_paths
+    get_installed_distributions.return_value = [fake_dist("django==1.8")]
+
+    runner.invoke(cli, ["--python-executable", custom_executable])
+
+    get_installed_distributions.assert_called_once_with(
+        skip=[], user_only=False, paths=sys_paths, local_only=False
+    )
+
+    assert run.call_count == 2
+
+    call_args = [call[0][0] for call in run.call_args_list]
+    called_uninstall_options = [args for args in call_args if args[3] == "uninstall"]
+    called_install_options = [args[:-1] for args in call_args if args[3] == "install"]
+
+    assert called_uninstall_options == [
+        [custom_executable, "-m", "pip", "uninstall", "-y", "django"]
+    ]
+    assert called_install_options == [[custom_executable, "-m", "pip", "install", "-r"]]
+
+
+def test_invalid_python_executable(runner):
+    with open("requirements.txt", "w") as req_in:
+        req_in.write("small-fake-a==1.10.0")
+
+    out = runner.invoke(cli, ["--python-executable", "/tmp/invalid_executable"])
+    assert out.exit_code == 2, out
+
+
+@mock.patch("piptools.sync.run")
+def test_default_python_executable_option(run, runner):
+    """
+    Make sure sys.executable is used when --python-executable is not provided
+    """
+    with open("requirements.txt", "w") as req_in:
+        req_in.write("small-fake-a==1.10.0")
+
+    runner.invoke(cli)
+
+    assert run.call_count == 2
+
+    call_args = [call[0][0] for call in run.call_args_list]
+    called_install_options = [args[:-1] for args in call_args if args[3] == "install"]
+    assert called_install_options == [
+        [
+            sys.executable,
+            "-m",
+            "pip",
+            "install",
+            "-r",
+        ]
+    ]

tests/test_utils.py

@@ -2,6 +2,7 @@
 import operator
 import os
 import shlex
+import sys
 
 import pytest
 
@@ -14,6 +15,7 @@
     format_specifier,
     get_compile_command,
     get_hashes_from_ireq,
+    get_sys_path_for_python_executable,
     is_pinned_requirement,
     is_url_requirement,
     lookup_table,
@@ -367,3 +369,10 @@ def test_lookup_table_from_tuples_with_empty_values():
 
 def test_lookup_table_with_empty_values():
     assert lookup_table((), operator.itemgetter(0)) == {}
+
+
+def test_get_sys_path_for_python_executable():
+    result = get_sys_path_for_python_executable(sys.executable)
+    # not testing for equality, because pytest adds extra paths into current sys.path
+    for path in result:
+        assert path in sys.path