WIP: Refactoring install command

condax/cli/install.py

@@ -1,11 +1,8 @@
 import logging
-from typing import List
+from pathlib import Path
+from typing import Iterable, List
 
-import click
-
-import condax.config as config
-import condax.core as core
-from condax import __version__
+from condax import __version__, consts, core
 
 from . import cli, options
 
@@ -15,7 +12,7 @@
     Install a package with condax.
 
     This will install a package into a new conda environment and link the executable
-    provided by it to `{config.DEFAULT_BIN_DIR}`.
+    provided by it to `{consts.DEFAULT_PATHS.bin_dir}`.
     """
 )
 @options.channels
@@ -26,9 +23,17 @@ def install(
     packages: List[str],
     is_forcing: bool,
     log_level: int,
+    channels: Iterable[str],
+    prefix_dir: Path,
+    bin_dir: Path,
     **_,
 ):
     for pkg in packages:
         core.install_package(
-            pkg, is_forcing=is_forcing, conda_stdout=log_level <= logging.INFO
+            pkg,
+            is_forcing=is_forcing,
+            conda_stdout=log_level <= logging.INFO,
+            channels=channels,
+            location=prefix_dir,
+            bin_dir=bin_dir,
         )

condax/cli/repair.py

@@ -27,5 +27,5 @@
 def repair(is_migrating, **_):
     if is_migrating:
         migrate.from_old_version()
-    conda.setup_micromamba()
+    conda.install_micromamba()
     core.fix_links()

condax/conda.py

@@ -1,3 +1,4 @@
+from functools import partial
 import io
 import json
 import logging
@@ -15,7 +16,7 @@
 
 from condax.config import C
 from condax.exceptions import CondaxError
-from condax.utils import to_path
+from condax.utils import FullPath
 import condax.utils as utils
 
 
@@ -26,36 +27,36 @@ def _ensure(execs: Iterable[str], installer: Callable[[], Path]) -> Path:
     for exe in execs:
         exe_path = shutil.which(exe)
         if exe_path is not None:
-            return to_path(exe_path)
+            return FullPath(exe_path)
 
     logger.info("No existing conda installation found. Installing the standalone")
     return installer()
 
 
-def ensure_conda() -> Path:
-    return _ensure(("conda", "mamba"), setup_conda)
+def ensure_conda(bin_dir: Path) -> Path:
+    return _ensure(("conda", "mamba"), partial(install_conda, bin_dir))
 
 
-def ensure_micromamba() -> Path:
-    return _ensure(("micromamba",), setup_micromamba)
+def ensure_micromamba(bin_dir: Path) -> Path:
+    return _ensure(("micromamba",), partial(install_micromamba, bin_dir))
 
 
-def setup_conda() -> Path:
+def install_conda(bin_dir: Path) -> Path:
     url = utils.get_conda_url()
     resp = requests.get(url, allow_redirects=True)
     resp.raise_for_status()
-    utils.mkdir(C.bin_dir())
+    utils.mkdir(bin_dir)
     exe_name = "conda.exe" if os.name == "nt" else "conda"
-    target_filename = C.bin_dir() / exe_name
+    target_filename = bin_dir / exe_name
     with open(target_filename, "wb") as fo:
         fo.write(resp.content)
     st = os.stat(target_filename)
     os.chmod(target_filename, st.st_mode | stat.S_IXUSR)
     return target_filename
 
 
-def setup_micromamba() -> Path:
-    utils.mkdir(C.bin_dir())
+def install_micromamba(bin_dir: Path) -> Path:
+    utils.mkdir(bin_dir)
     exe_name = "micromamba.exe" if os.name == "nt" else "micromamba"
     umamba_exe = C.bin_dir() / exe_name
     _download_extract_micromamba(umamba_exe)
@@ -89,16 +90,20 @@ def _download_extract_micromamba(umamba_dst: Path) -> None:
 #     )
 
 
-def create_conda_environment(spec: str, stdout: bool) -> None:
+def create_conda_environment(
+    prefix: Path, spec: str, stdout: bool, channels: Iterable[str], bin_dir: Path
+) -> None:
     """Create an environment by installing a package.
 
     NOTE: `spec` may contain version specificaitons.
-    """
-    conda_exe = ensure_conda()
-    prefix = conda_env_prefix(spec)
 
-    channels = C.channels()
-    channels_args = [x for c in channels for x in ["--channel", c]]
+    Args:
+        spec: Package spec to install. e.g. "python=3.6", "python>=3.6", "python", etc.
+        stdout: Whether to print stdout.
+        location: Directory in which to create the environment.
+    """
+    conda_exe = ensure_conda(bin_dir)
+    channels_args = (x for c in channels for x in ("--channel", c))
 
     _subprocess_run(
         [
@@ -163,12 +168,12 @@ def uninject_from_conda_env(
     )
 
 
-def remove_conda_env(package: str, stdout: bool) -> None:
+def remove_conda_env(env: Path, stdout: bool) -> None:
     """Remove a conda environment."""
     conda_exe = ensure_conda()
 
     _subprocess_run(
-        [conda_exe, "remove", "--prefix", conda_env_prefix(package), "--all", "--yes"],
+        [conda_exe, "remove", "--prefix", env, "--all", "--yes"],
         suppress_stdout=not stdout,
     )
 
@@ -212,15 +217,9 @@ def update_conda_env(spec: str, update_specs: bool, stdout: bool) -> None:
     _subprocess_run(command, suppress_stdout=not stdout)
 
 
-def has_conda_env(package: str) -> bool:
+def is_conda_env(path: Path) -> bool:
     # TODO: check some properties of a conda environment
-    p = conda_env_prefix(package)
-    return p.exists() and p.is_dir()
-
-
-def conda_env_prefix(spec: str) -> Path:
-    package, _ = utils.split_match_specs(spec)
-    return C.prefix_dir() / package
+    return path.is_dir()
 
 
 def get_package_info(package: str, specific_name=None) -> Tuple[str, str, str]:
@@ -251,37 +250,33 @@ def __init__(self, package: str):
 
 
 def determine_executables_from_env(
-    package: str, injected_package: Optional[str] = None
+    prefix: Path, package: str, injected_package: Optional[str] = None
 ) -> List[Path]:
-    def is_good(p: Union[str, Path]) -> bool:
-        p = to_path(p)
-        return p.parent.name in ("bin", "sbin", "scripts", "Scripts")
+    def is_exe(p: Union[str, Path]) -> bool:
+        return FullPath(p).parent.name in ("bin", "sbin", "scripts", "Scripts")
 
-    env_prefix = conda_env_prefix(package)
     target_name = injected_package if injected_package else package
 
-    conda_meta_dir = env_prefix / "conda-meta"
+    conda_meta_dir = prefix / "conda-meta"
     for file_name in conda_meta_dir.glob(f"{target_name}*.json"):
         with file_name.open() as fo:
             package_info = json.load(fo)
             if package_info["name"] == target_name:
                 potential_executables: Set[str] = {
                     fn
                     for fn in package_info["files"]
-                    if (fn.startswith("bin/") and is_good(fn))
-                    or (fn.startswith("sbin/") and is_good(fn))
+                    if (fn.startswith("bin/") and is_exe(fn))
+                    or (fn.startswith("sbin/") and is_exe(fn))
                     # They are Windows style path
-                    or (fn.lower().startswith("scripts") and is_good(fn))
-                    or (fn.lower().startswith("library") and is_good(fn))
+                    or (fn.lower().startswith("scripts") and is_exe(fn))
+                    or (fn.lower().startswith("library") and is_exe(fn))
                 }
                 break
     else:
         raise DeterminePkgFilesError(target_name)
 
     return sorted(
-        env_prefix / fn
-        for fn in potential_executables
-        if utils.is_executable(env_prefix / fn)
+        prefix / fn for fn in potential_executables if utils.is_executable(prefix / fn)
     )
 
 
@@ -297,7 +292,7 @@ def _get_conda_package_dirs() -> List[Path]:
         return []
 
     d = json.loads(res.stdout.decode())
-    return [to_path(p) for p in d["pkgs_dirs"]]
+    return [FullPath(p) for p in d["pkgs_dirs"]]
 
 
 def _get_dependencies(package: str, pkg_dir: Path) -> List[str]:

condax/config.py

@@ -4,7 +4,7 @@
 from typing import Any, Dict, List, Optional, Union
 from condax.exceptions import CondaxError
 
-from condax.utils import to_path
+from condax.utils import FullPath
 import condax.condarc as condarc
 import yaml
 
@@ -17,30 +17,30 @@
     _localappdata_dir, "condax", "condax", _config_filename
 )
 _default_config = _default_config_windows if os.name == "nt" else _default_config_unix
-DEFAULT_CONFIG = to_path(os.environ.get("CONDAX_CONFIG", _default_config))
+DEFAULT_CONFIG = FullPath(os.environ.get("CONDAX_CONFIG", _default_config))
 
 _xdg_data_home = os.environ.get("XDG_DATA_HOME", "~/.local/share")
 _default_prefix_dir_unix = os.path.join(_xdg_data_home, "condax", "envs")
 _default_prefix_dir_win = os.path.join(_localappdata_dir, "condax", "condax", "envs")
 _default_prefix_dir = (
     _default_prefix_dir_win if os.name == "nt" else _default_prefix_dir_unix
 )
-DEFAULT_PREFIX_DIR = to_path(os.environ.get("CONDAX_PREFIX_DIR", _default_prefix_dir))
+DEFAULT_PREFIX_DIR = FullPath(os.environ.get("CONDAX_PREFIX_DIR", _default_prefix_dir))
 
-DEFAULT_BIN_DIR = to_path(os.environ.get("CONDAX_BIN_DIR", "~/.local/bin"))
+DEFAULT_BIN_DIR = FullPath(os.environ.get("CONDAX_BIN_DIR", "~/.local/bin"))
 
 _channels_in_condarc = condarc.load_channels()
 DEFAULT_CHANNELS = (
     os.environ.get("CONDAX_CHANNELS", " ".join(_channels_in_condarc)).strip().split()
 )
 
-CONDA_ENVIRONMENT_FILE = to_path("~/.conda/environments.txt")
+CONDA_ENVIRONMENT_FILE = FullPath("~/.conda/environments.txt")
 
 conda_path = shutil.which("conda")
 MAMBA_ROOT_PREFIX = (
-    to_path(conda_path).parent.parent
+    FullPath(conda_path).parent.parent
     if conda_path is not None
-    else to_path(os.environ.get("MAMBA_ROOT_PREFIX", "~/micromamba"))
+    else FullPath(os.environ.get("MAMBA_ROOT_PREFIX", "~/micromamba"))
 )
 
 
@@ -94,7 +94,7 @@ def set_via_file(config_file: Union[str, Path]):
     Raises:
         BadConfigFileError: If the config file is not valid.
     """
-    config_file = to_path(config_file)
+    config_file = FullPath(config_file)
     try:
         with config_file.open() as f:
             config = yaml.safe_load(f)
@@ -107,20 +107,20 @@ def set_via_file(config_file: Union[str, Path]):
 
     # For compatibility with condax 0.0.5
     if "prefix_path" in config:
-        prefix_dir = to_path(config["prefix_path"])
+        prefix_dir = FullPath(config["prefix_path"])
         C._set("prefix_dir", prefix_dir)
 
     # For compatibility with condax 0.0.5
     if "target_destination" in config:
-        bin_dir = to_path(config["target_destination"])
+        bin_dir = FullPath(config["target_destination"])
         C._set("bin_dir", bin_dir)
 
     if "prefix_dir" in config:
-        prefix_dir = to_path(config["prefix_dir"])
+        prefix_dir = FullPath(config["prefix_dir"])
         C._set("prefix_dir", prefix_dir)
 
     if "bin_dir" in config:
-        bin_dir = to_path(config["bin_dir"])
+        bin_dir = FullPath(config["bin_dir"])
         C._set("bin_dir", bin_dir)
 
     if "channels" in config:
@@ -137,10 +137,10 @@ def set_via_value(
     Set a part of values in the object C by passing values directly.
     """
     if prefix_dir:
-        C._set("prefix_dir", to_path(prefix_dir))
+        C._set("prefix_dir", FullPath(prefix_dir))
 
     if bin_dir:
-        C._set("bin_dir", to_path(bin_dir))
+        C._set("bin_dir", FullPath(bin_dir))
 
     if channels:
         C._set("channels", channels + C.channels())

condax/consts.py

@@ -0,0 +1,54 @@
+from dataclasses import dataclass
+import os
+from pathlib import Path
+
+
+from condax.utils import FullPath
+
+
+IS_WIN = os.name == "nt"
+IS_UNIX = not IS_WIN
+
+
+@dataclass
+class Paths:
+    conf_dir: Path
+    bin_dir: Path
+    data_dir: Path
+    conf_file_name: str = "condax.yml"
+    envs_dir_name: str = "envs"
+
+    @property
+    def conf_file(self) -> Path:
+        return self.conf_dir / self.conf_file_name
+
+    @property
+    def prefix_dir(self) -> Path:
+        return self.data_dir / self.envs_dir_name
+
+
+class _WindowsPaths(Paths):
+    def __init__(self):
+        conf_dir = data_dir = (
+            FullPath(os.environ.get("LOCALAPPDATA", "~/AppData/Local"))
+            / "condax/condax"
+        )
+        super().__init__(
+            conf_dir=conf_dir,
+            bin_dir=conf_dir / "bin",
+            data_dir=data_dir,
+        )
+
+
+class _UnixPaths(Paths):
+    def __init__(self):
+        super().__init__(
+            conf_dir=FullPath(os.environ.get("XDG_CONFIG_HOME", "~/.config"))
+            / "condax",
+            bin_dir=FullPath("~/.local/bin"),
+            data_dir=FullPath(os.environ.get("XDG_DATA_HOME", "~/.local/share"))
+            / "condax",
+        )
+
+
+DEFAULT_PATHS: Paths = _UnixPaths() if IS_UNIX else _WindowsPaths()