WIP: Refactoring remove command

condax/cli/options.py

@@ -22,6 +22,7 @@ def common(f: Callable) -> Callable:
     """
     options: Sequence[Callable] = (
         condax,
+        log_level,
         click.help_option("-h", "--help"),
     )
 
@@ -108,23 +109,17 @@ def _config_file_callback(_, __, config_file: Path) -> Mapping[str, Any]:
 
 def conda(f: Callable) -> Callable:
     """
-    This click option decorator adds the --channel and --config options as well as all those added by `options.log_level` to the CLI.
+    This click option decorator adds the --channel and --config options to the CLI.
     It constructs a `Conda` object and passes it to the decorated function as `conda`.
     It reads the config file and passes it as a dict to the decorated function as `config`.
     """
 
-    @log_level
     @config
     @wraps(f)
-    def construct_conda_hook(config: Mapping[str, Any], log_level: int, **kwargs):
+    def construct_conda_hook(config: Mapping[str, Any], **kwargs):
         return f(
-            conda=Conda(
-                config.get("channels", []),
-                stdout=subprocess.DEVNULL if log_level >= logging.INFO else None,
-                stderr=subprocess.DEVNULL if log_level >= logging.CRITICAL else None,
-            ),
+            conda=Conda(config.get("channels", [])),
             config=config,
-            log_level=log_level,
             **kwargs,
         )
 

condax/cli/remove.py

@@ -1,6 +1,6 @@
 import logging
 from typing import List
-import click
+from condax.condax import Condax
 
 import condax.core as core
 from condax import __version__
@@ -18,9 +18,9 @@
 )
 @options.common
 @options.packages
-def remove(packages: List[str], log_level: int, **_):
+def remove(packages: List[str], condax: Condax, **_):
     for pkg in packages:
-        core.remove_package(pkg, conda_stdout=log_level <= logging.INFO)
+        condax.remove_package(pkg)
 
 
 @cli.command(

condax/conda/conda.py

@@ -3,7 +3,9 @@
 import shlex
 import subprocess
 import logging
-from typing import Iterable
+import sys
+from typing import IO, Iterable, Optional
+from halo import Halo
 
 from condax import consts
 from .installers import ensure_conda
@@ -13,35 +15,26 @@
 
 
 class Conda:
-    def __init__(
-        self,
-        channels: Iterable[str],
-        stdout=subprocess.DEVNULL,
-        stderr=None,
-    ) -> None:
+    def __init__(self, channels: Iterable[str]) -> None:
         """This class is a wrapper for conda's CLI.
 
         Args:
             channels: Additional channels to use.
-            stdout (optional): This is passed directly to `subprocess.run`. Defaults to subprocess.DEVNULL.
-            stderr (optional): This is passed directly to `subprocess.run`. Defaults to None.
         """
         self.channels = tuple(channels)
-        self.stdout = stdout
-        self.stderr = stderr
         self.exe = ensure_conda(consts.DEFAULT_PATHS.bin_dir)
 
-    @classmethod
-    def is_env(cls, path: Path) -> bool:
-        return (path / "conda-meta").is_dir()
-
     def remove_env(self, env: Path) -> None:
         """Remove a conda environment.
 
         Args:
             env: The path to the environment to remove.
         """
-        self._run(f"remove --prefix {env} --all --yes")
+        self._run(
+            f"env remove --prefix {env} --yes",
+            stdout_level=logging.DEBUG,
+            stderr_level=logging.INFO,
+        )
 
     def create_env(
         self,
@@ -58,21 +51,63 @@ def create_env(
             spec: Package spec to install. e.g. "python=3.6", "python>=3.6", "python", etc.
             extra_channels: Additional channels to search for packages in.
         """
-        self._run(
-            f"create --prefix {prefix} {' '.join(f'--channel {c}' for c in itertools.chain(extra_channels, self.channels))} --quiet --yes {shlex.quote(spec)}"
-        )
+        cmd = f"create --prefix {prefix} {' '.join(f'--channel {c}' for c in itertools.chain(extra_channels, self.channels))} --quiet --yes {shlex.quote(spec)}"
+        if logger.getEffectiveLevel() <= logging.INFO:
+            with Halo(
+                text=f"Creating environment for {spec}",
+                spinner="dots",
+                stream=sys.stderr,
+            ):
+                self._run(cmd)
+        else:
+            self._run(cmd)
 
-    def _run(self, command: str) -> subprocess.CompletedProcess:
+    def _run(
+        self,
+        command: str,
+        stdout_level: int = logging.DEBUG,
+        stderr_level: int = logging.ERROR,
+    ) -> subprocess.CompletedProcess:
         """Run a conda command.
 
         Args:
             command: The command to run excluding the conda executable.
         """
-        cmd = shlex.split(f"{self.exe} {command}")
+        cmd = f"{self.exe} {command}"
         logger.debug(f"Running: {cmd}")
-        return subprocess.run(
-            cmd,
-            stdout=self.stdout,
-            stderr=self.stderr,
-            text=True,
+        cmd_list = shlex.split(cmd)
+
+        p = subprocess.Popen(
+            cmd_list, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
         )
+
+        stdout_done, stderr_done = False, False
+        while not stdout_done or not stderr_done:
+            stdout_done = self._log_stream(p.stdout, stdout_level)
+            stderr_done = self._log_stream(p.stderr, stderr_level)
+
+        ret_code = p.wait()
+
+        return subprocess.CompletedProcess(
+            cmd_list,
+            ret_code,
+            p.stdout.read() if p.stdout else None,
+            p.stderr.read() if p.stderr else None,
+        )
+
+    def _log_stream(self, stream: Optional[IO[str]], log_level: int) -> bool:
+        """Log one line of process ouput.
+
+        Args:
+            stream: The stream to read from.
+            log_level: The log level to use.
+
+        Returns:
+            True if the stream is depleted. False otherwise.
+        """
+        if stream is None:
+            return True
+        line = stream.readline()
+        if line:
+            logger.log(log_level, f"\r{line.rstrip()}")
+        return not line

condax/conda/env_info.py

@@ -7,6 +7,10 @@
 from .exceptions import NoPackageMetadata
 
 
+def is_env(path: Path) -> bool:
+    return (path / "conda-meta").is_dir()
+
+
 def find_exes(prefix: Path, package: str) -> List[Path]:
     """Find executables in environment `prefix` provided py a given `package`.
 

condax/condax/condax.py

@@ -5,7 +5,9 @@
 from condax import utils
 from condax.conda import Conda, env_info
 from .exceptions import PackageInstalledError, NotAnEnvError
-from . import links, metadata
+
+from . import links
+from .metadata import metadata
 
 logger = logging.getLogger(__name__)
 
@@ -38,7 +40,7 @@ def install_package(
         package = utils.package_name(spec)
         env = self.prefix_dir / package
 
-        if self.conda.is_env(env):
+        if env_info.is_env(env):
             if is_forcing:
                 logger.warning(f"Overwriting environment for {package}")
                 self.conda.remove_env(env)
@@ -53,3 +55,14 @@ def install_package(
         links.create_links(env, executables, self.bin_dir, is_forcing=is_forcing)
         metadata.create_metadata(env, package, executables)
         logger.info(f"`{package}` has been installed by condax")
+
+    def remove_package(self, package: str):
+        env = self.prefix_dir / package
+        if not env_info.is_env(env):
+            logger.warning(f"{package} is not installed with condax")
+            return
+
+        apps_to_unlink = metadata.load(env).apps
+        links.remove_links(package, self.bin_dir, apps_to_unlink)
+        self.conda.remove_env(env)
+        logger.info(f"`{package}` has been removed from condax")

condax/condax/exceptions.py

@@ -18,8 +18,6 @@ def __init__(self, location: Path, msg: str = ""):
         )
 
 
-class BadMetadataError(CondaxError):
-    def __init__(self, metadata_path: Path, msg: str):
-        super().__init__(
-            103, f"Error loading condax metadata at {metadata_path}: {msg}"
-        )
+class PackageNotInstalled(CondaxError):
+    def __init__(self, package: str):
+        super().__init__(103, f"Package `{package}` is not installed with condax")

condax/condax/links.py

@@ -2,10 +2,10 @@
 import os
 from pathlib import Path
 import shutil
-from typing import Iterable
+from typing import Iterable, List
 
 from condax.conda import installers
-from condax import utils
+from condax import utils, wrapper
 
 logger = logging.getLogger(__name__)
 
@@ -49,7 +49,7 @@ def create_link(env: Path, exe: Path, location: Path, is_forcing: bool = False)
     if os.name == "nt":
         script_lines = [
             "@rem Entrypoint created by condax\n",
-            f"@call {utils.quote(micromamba_exe)} run --prefix {utils.quote(env)} {utils.quote(exe)} %*\n",
+            f'@call "{micromamba_exe}" run --prefix "{env}" "{exe}" %*\n',
         ]
     else:
         script_lines = [
@@ -78,6 +78,37 @@ def create_link(env: Path, exe: Path, location: Path, is_forcing: bool = False)
     return True
 
 
+def remove_links(package: str, location: Path, executables_to_unlink: Iterable[str]):
+    unlinked: List[str] = []
+    for executable_name in executables_to_unlink:
+        link_path = location / _get_wrapper_name(executable_name)
+        if os.name == "nt":
+            # FIXME: this is hand-waving for now
+            utils.unlink(link_path)
+        else:
+            wrapper_env = wrapper.read_env_name(link_path)
+
+            if wrapper_env is None:
+                utils.unlink(link_path)
+                unlinked.append(f"{executable_name} \t (failed to get env)")
+                continue
+
+            if wrapper_env != package:
+                logger.warning(
+                    f"Keeping {executable_name} as it runs in environment `{wrapper_env}`, not `{package}`."
+                )
+                continue
+
+            link_path.unlink()
+
+        unlinked.append(executable_name)
+
+    if executables_to_unlink:
+        logger.info(
+            "\n  - ".join(("Removed the following entrypoint links:", *unlinked))
+        )
+
+
 def _get_wrapper_name(name: str) -> str:
     """Get the file name of the entrypoint script for the executable with the given name.