merge ManagerInfo in BaseManager

Author: woutdenolf

src/ewoks/_requirements/managers/conda.py

@@ -1,4 +1,8 @@
 import logging
+import os
+import sys
+from typing import Optional
+from typing import Tuple
 
 import yaml
 
@@ -11,6 +15,26 @@
 
 class CondaManager(BaseManager):
     NAME = "conda"
+    PRIORITY = 4
+
+    def __init__(self, *command: str) -> None:
+        if not command:
+            command = self._get_conda_command()
+        super().__init__(*command)
+
+    def version(self) -> Optional[str]:
+        """Returns None when this manager is not available."""
+        try:
+            output = self._check_output("--version")
+        except RuntimeError:
+            return None
+        return output.strip().split(" ")[-1]
+
+    def is_active(self) -> bool:
+        """Manager is explicitly active."""
+        return "CONDA_PREFIX" in os.environ or os.path.exists(
+            os.path.join(sys.prefix, "conda-meta")
+        )
 
     def _gather_requirements(self, manager_version: str) -> CondaRequirements:
         output = self._check_output("env", "export")
@@ -28,3 +52,16 @@ def install_requirements(self, requirements: CondaRequirements) -> None:
         text = yaml.safe_dump(requirements.environment)
         with self._temporary_file(text, ".yml") as tmp_path:
             self._check_call("env", "update", "-f", tmp_path)
+
+    def _get_conda_command(self) -> Tuple[str, ...]:
+        try:
+            _ = self._check_output_raw("mamba", "--version")
+            return ("mamba",)
+        except Exception:
+            pass
+        try:
+            _ = self._check_output_raw("micromamba", "--version")
+            return ("micromamba",)
+        except Exception:
+            pass
+        return ("conda",)

src/ewoks/_requirements/managers/pip.py

@@ -1,5 +1,8 @@
+import importlib.metadata
 import logging
+import sys
 from typing import List
+from typing import Optional
 
 from ..metadata import pip_freeze
 from ..metadata.gather import gather_requirements
@@ -11,6 +14,23 @@
 
 class PipManager(BaseManager):
     NAME = "pip"
+    PRIORITY = 0
+
+    def __init__(self, *command: str) -> None:
+        if not command:
+            command = sys.executable, "-m", "pip"
+        super().__init__(*command)
+
+    def version(self) -> Optional[str]:
+        """Returns None when this manager is not available."""
+        try:
+            return importlib.metadata.version("pip")
+        except importlib.metadata.PackageNotFoundError:
+            return None
+
+    def is_active(self) -> bool:
+        """Manager is explicitly active."""
+        return False
 
     def _gather_requirements(self, manager_version: str) -> PipRequirements:
         freeze_output = self._check_output("freeze").strip().splitlines()

src/ewoks/_requirements/managers/pipenv.py

@@ -1,4 +1,8 @@
+import importlib.metadata
 import json
+import os
+import sys
+from typing import Optional
 
 from ..metadata.gather import gather_requirements
 from ..models.pipenv import PipenvRequirements
@@ -7,6 +11,23 @@
 
 class PipenvManager(BaseManager):
     NAME = "pipenv"
+    PRIORITY = 3
+
+    def __init__(self, *command: str) -> None:
+        if not command:
+            command = sys.executable, "-m", "pipenv"
+        super().__init__(*command)
+
+    def version(self) -> Optional[str]:
+        """Returns None when this manager is not available."""
+        try:
+            return importlib.metadata.version("pipenv")
+        except importlib.metadata.PackageNotFoundError:
+            return None
+
+    def is_active(self) -> bool:
+        """Manager is explicitly active."""
+        return "PIPENV_ACTIVE" in os.environ
 
     def _gather_requirements(self, manager_version: str) -> PipenvRequirements:
         output = self._check_output("lock", "--requirements")

src/ewoks/_requirements/managers/pixi.py

@@ -1,4 +1,5 @@
 import os
+from typing import Optional
 
 from ..metadata.gather import gather_requirements
 from ..models.pixi import PixiRequirements
@@ -7,6 +8,24 @@
 
 class PixiManager(BaseManager):
     NAME = "pixi"
+    PRIORITY = 5
+
+    def __init__(self, *command: str) -> None:
+        if not command:
+            command = ("pixi",)
+        super().__init__(*command)
+
+    def version(self) -> Optional[str]:
+        """Returns None when this manager is not available."""
+        try:
+            output = self._check_output("--version", text=True)
+            return output.strip().split(" ")[-1]
+        except Exception:
+            return None
+
+    def is_active(self) -> bool:
+        """Manager is explicitly active."""
+        return "PIXI_PROJECT_ROOT" in os.environ
 
     def _gather_requirements(self, manager_version: str) -> PixiRequirements:
         if os.path.exists("pixi.lock"):

src/ewoks/_requirements/managers/poetry.py

@@ -1,10 +1,32 @@
+import importlib.metadata
+import os
+import sys
+from typing import Optional
+
 from ..metadata.gather import gather_requirements
 from ..models.poetry import PoetryRequirements
 from .utils.base import BaseManager
 
 
 class PoetryManager(BaseManager):
     NAME = "poetry"
+    PRIORITY = 2
+
+    def __init__(self, *command: str) -> None:
+        if not command:
+            command = sys.executable, "-m", "poetry"
+        super().__init__(*command)
+
+    def version(self) -> Optional[str]:
+        """Returns None when this manager is not available."""
+        try:
+            return importlib.metadata.version("poetry")
+        except importlib.metadata.PackageNotFoundError:
+            return None
+
+    def is_active(self) -> bool:
+        """Manager is explicitly active."""
+        return "POETRY_ACTIVE" in os.environ
 
     def _gather_requirements(self, manager_version: str) -> PoetryRequirements:
         output = self._check_output("export", "--without-hashes")

src/ewoks/_requirements/managers/utils/base.py

@@ -5,11 +5,9 @@
 from abc import abstractmethod
 from contextlib import contextmanager
 from typing import Generator
-from typing import List
 from typing import Optional
 
 from ...models.base import BaseRequirements
-from .commands import get_manager_command
 
 logger = logging.getLogger(__name__)
 
@@ -34,18 +32,17 @@ class BaseManager:
     """
 
     NAME = NotImplemented
+    PRIORITY = NotImplemented
 
     def __init__(self, *command: str) -> None:
         if not command:
-            command = get_manager_command(self.NAME)
+            raise ValueError(f"{type(self).__name__} needs an associated shell command")
         self._cmd_args = command
 
     def gather_requirements(self) -> Optional[BaseRequirements]:
         """Return requirements generated from the current python environment."""
-        from .supported import get_supported_managers
-
-        manager_version = get_supported_managers()[self.NAME].version
-        if not manager_version:
+        manager_version = self.current_version()
+        if manager_version is None:
             raise RuntimeError(f"{self.NAME!r} is not installed")
 
         try:
@@ -56,6 +53,16 @@ def gather_requirements(self) -> Optional[BaseRequirements]:
             )
             return None
 
+    @abstractmethod
+    def version(self) -> Optional[str]:
+        """Returns None when this manager is not available."""
+        pass
+
+    @abstractmethod
+    def is_active(self) -> bool:
+        """Manager is explicitly active."""
+        pass
+
     def install_requirements(self, requirements: BaseRequirements) -> None:
         """Install requirements into the current python environment."""
         try:
@@ -74,13 +81,25 @@ def _gather_requirements(self, manager_version: str) -> BaseRequirements:
     def _install_requirements(self, requirements: BaseRequirements) -> None:
         pass
 
-    def _check_output(self, *args) -> str:
-        return _check_output([*self._cmd_args, *args])
+    def _check_output(self, *args: str) -> str:
+        return self._check_output_raw(*[*self._cmd_args, *args])
 
-    def _check_call(self, *args, raw: bool = False) -> int:
-        if raw:
-            return _check_call([*args])
-        return _check_call([*self._cmd_args, *args])
+    def _check_call(self, *args: str) -> int:
+        return self._check_call_raw(*[*self._cmd_args, *args])
+
+    @staticmethod
+    def _check_output_raw(*args: str) -> str:
+        try:
+            return subprocess.check_output(args, text=True)
+        except Exception as ex:
+            raise RuntimeError(f"Command failed: {args}") from ex
+
+    @staticmethod
+    def _check_call_raw(*args: str) -> int:
+        try:
+            return subprocess.check_call(args)
+        except Exception as ex:
+            raise RuntimeError(f"Command failed: {args}") from ex
 
     @contextmanager
     def _temporary_file(self, text: str, suffix: str) -> Generator[str, None, None]:
@@ -98,17 +117,3 @@ def _temporary_file(self, text: str, suffix: str) -> Generator[str, None, None]:
                     os.remove(tmp_path)
                 except OSError:
                     logger.debug("Could not delete temporary file: %s", tmp_path)
-
-
-def _check_output(args: List[str]) -> str:
-    try:
-        return subprocess.check_output(args, text=True)
-    except Exception as ex:
-        raise RuntimeError(f"Command failed: {args}") from ex
-
-
-def _check_call(args: List[str]) -> int:
-    try:
-        return subprocess.check_call(args)
-    except Exception as ex:
-        raise RuntimeError(f"Command failed: {args}") from ex