Add dependecy on ensureconda instead of duplicating that functionality

conda_lock/conda_lock.py

@@ -4,33 +4,17 @@
 
 import argparse
 import atexit
-import contextlib
 import json
-import logging
 import os
 import pathlib
-import platform
 import shutil
-import stat
 import subprocess
 import sys
 import tempfile
 
-from typing import (
-    IO,
-    Dict,
-    Iterable,
-    Iterator,
-    List,
-    MutableSequence,
-    Optional,
-    Sequence,
-    Set,
-    Tuple,
-    Union,
-)
-
-import requests
+from typing import Dict, List, MutableSequence, Optional, Sequence, Set, Tuple, Union
+
+import ensureconda
 
 from conda_lock.src_parser import LockSpecification
 from conda_lock.src_parser.environment_yaml import parse_environment_file
@@ -46,95 +30,8 @@
     sys.exit(1)
 
 
-DEFAULT_PLATFORMS = ["osx-64", "linux-64", "win-64"]
-
-
-def candidate_executables(
-    conda_executable: Optional[str], no_mamba: bool
-) -> Iterable[Optional[str]]:
-    if conda_executable:
-        if pathlib.Path(conda_executable).exists():
-            yield conda_executable
-        yield shutil.which(conda_executable)
-    if not no_mamba:
-        yield shutil.which("mamba")
-        # micromamba doesn't support dry-run yet
-        # yield shutil.which("micromamba")
-        # yield shutil.which("micromamba.exe")
-    yield shutil.which("conda")
-    yield shutil.which("conda.exe")
-
-
-def ensure_conda(
-    conda_executable: Optional[str] = None, no_mamba: bool = False
-) -> pathlib.Path:
-    for candidate in candidate_executables(conda_executable, no_mamba):
-        if candidate:
-            return pathlib.Path(candidate)
-    else:
-        logging.info(
-            "No existing conda installation found.  Installing the standalone conda solver"
-        )
-        return pathlib.Path(install_conda_exe())
-
-
-@contextlib.contextmanager
-def new_executable(target_filename: PathLike) -> Iterator[IO[bytes]]:
-    with open(target_filename, "wb") as fo:
-        yield fo
-    st = os.stat(target_filename)
-    os.chmod(target_filename, st.st_mode | stat.S_IXUSR)
-
-
-def install_conda_exe() -> str:
-    conda_exe_prefix = "https://repo.anaconda.com/pkgs/misc/conda-execs"
-    if platform.system() == "Linux":
-        conda_exe_file = "conda-latest-linux-64.exe"
-    elif platform.system() == "Darwin":
-        conda_exe_file = "conda-latest-osx-64.exe"
-    elif platform.system() == "NT":
-        conda_exe_file = "conda-latest-win-64.exe"
-    else:
-        # TODO: Support windows here
-        raise ValueError(f"Unsupported platform: {platform.system()}")
-
-    resp = requests.get(f"{conda_exe_prefix}/{conda_exe_file}", allow_redirects=True)
-    resp.raise_for_status()
-    target_filename = os.path.expanduser(pathlib.Path(__file__).parent / "conda.exe")
-    with new_executable(target_filename) as fo:
-        fo.write(resp.content)
-    return target_filename
-
-
-def install_micromamba_exe() -> str:
-    """Install micromamba into the conda-lock installation"""
-    if platform.system() == "Linux":
-        p = "linux-64"
-    elif platform.system() == "Darwin":
-        p = "osx-64"
-    else:
-        raise ValueError(f"Unsupported platform: {platform.system()}")
-    url = f"https://micromamba.snakepit.net/api/micromamba/{p}/latest"
-    resp = requests.get(url, allow_redirects=True)
-    resp.raise_for_status()
-    import io
-    import tarfile
-
-    tarball = io.BytesIO(resp.content)
-    tarball.seek(0)
-    with tarfile.open(mode="r:bz2", fileobj=tarball) as tf:
-        fo = tf.extractfile("bin/micromamba")
-        if fo is None:
-            raise RuntimeError("Could not extract micromamba executable!")
-        target_filename = os.path.expanduser(
-            pathlib.Path(__file__).parent / "micromamba.exe"
-        )
-        with new_executable(target_filename) as exe_fo:
-            exe_fo.write(fo.read())
-        return target_filename
-
-
 CONDA_PKGS_DIRS = None
+DEFAULT_PLATFORMS = ["osx-64", "linux-64", "win-64"]
 
 
 def conda_pkgs_dir():
@@ -449,6 +346,24 @@ def parse_source_file(
     return desired_env
 
 
+def _determine_conda_executable(conda_executable: Optional[str], no_mamba: bool):
+    if conda_executable:
+        if pathlib.Path(conda_executable).exists():
+            yield conda_executable
+        yield shutil.which(conda_executable)
+    _conda_exe = ensureconda.ensureconda(
+        mamba=not no_mamba, micromamba=not no_mamba, conda=True, conda_exe=True
+    )
+    yield _conda_exe
+
+
+def determine_conda_executable(conda_executable: Optional[str], no_mamba: bool):
+    for candidate in _determine_conda_executable(conda_executable, no_mamba):
+        if candidate is not None:
+            return candidate
+    raise RuntimeError("Could not find conda (or compatible) executable")
+
+
 def run_lock(
     environment_file: pathlib.Path,
     conda_exe: Optional[str],
@@ -457,7 +372,7 @@ def run_lock(
     include_dev_dependencies: bool = True,
     channel_overrides: Optional[Sequence[str]] = None,
 ) -> None:
-    _conda_exe = ensure_conda(conda_exe, no_mamba=no_mamba)
+    _conda_exe = determine_conda_executable(conda_exe, no_mamba=no_mamba)
     make_lock_files(
         conda=_conda_exe,
         src_file=environment_file,

requirements.txt

@@ -2,3 +2,4 @@ pyyaml
 requests
 jinja2
 toml
+ensureconda >=1.1

setup.cfg

@@ -32,6 +32,7 @@ install_requires =
     requests >=2
     Jinja2
     toml
+    ensureconda >=1.1
 python_requires = >=3.6
 packages = find:
 setup_requires =

tests/test_conda_lock.py

@@ -5,12 +5,7 @@
 
 import pytest
 
-from conda_lock.conda_lock import (
-    ensure_conda,
-    install_conda_exe,
-    parse_meta_yaml_file,
-    run_lock,
-)
+from conda_lock.conda_lock import parse_meta_yaml_file, run_lock
 from conda_lock.src_parser.environment_yaml import parse_environment_file
 from conda_lock.src_parser.pyproject_toml import (
     parse_flit_pyproject_toml,
@@ -58,20 +53,6 @@ def include_dev_dependencies(request: Any) -> bool:
     return request.param
 
 
-def test_ensure_conda_nopath():
-    assert pathlib.Path(ensure_conda()).is_file()
-
-
-def test_ensure_conda_path():
-    conda_executable = shutil.which("conda") or shutil.which("conda.exe")
-    assert pathlib.Path(conda_executable) == ensure_conda(conda_executable)
-
-
-def test_install_conda_exe():
-    target_filename = install_conda_exe()
-    assert pathlib.Path(target_filename) == ensure_conda(target_filename)
-
-
 def test_parse_environment_file(gdal_environment):
     res = parse_environment_file(gdal_environment, "linux-64")
     assert all(x in res.specs for x in ["python >=3.7,<3.8", "gdal"])