Support uv

CHANGELOG.md

@@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `ewoks install`: add `--package-manager-name` and `--package-manager-command` arguments.
 - `ewoks convert`: add `--package-manager-name` and `--package-manager-command` arguments.
 - Add `ewoks lint` CLI command to check if a workflow is conform to the Ewoks specification.
+- Support the `uv` package manager to generate and install workflow requirements.
 
 ### Removed
 

pyproject.toml

@@ -46,7 +46,7 @@ test = [
     "ipykernel",
     "importlib_resources",
     "pyqt5",
-    "pytest-venv",
+    "uv",
     "defusedxml",
 ]
 dev = [

src/ewoks/_requirements/__init__.py

@@ -1,20 +1,30 @@
 """Workflow requirements."""
 
 import logging
+from typing import Dict
 from typing import Optional
 from typing import Tuple
 from typing import Union
 
 from ewokscore.graph import TaskGraph
 
 from .utils import parse
+from .utils._supported import get_supported_managers
 from .utils.base_manager import BaseRequirements
 from .utils.detect import get_manager
 from .utils.metadata import last_resort
 
 logger = logging.getLogger(__file__)
 
 
+def supported_managers() -> Dict[str, str]:
+    """Names of the supported package managers with an example command for each."""
+    return {
+        name: manager_cls.COMMAND_EXAMPLE
+        for name, manager_cls in get_supported_managers().items()
+    }
+
+
 def add_requirements(
     graph: TaskGraph,
     manager_name: Optional[str] = None,

src/ewoks/_requirements/conda.py

@@ -29,6 +29,7 @@ class CondaManager(BaseManager):
     NAME = "conda"
     PRIORITY = 4
     REQUIREMENTS_MODEL = CondaRequirements
+    COMMAND_EXAMPLE = "conda"
 
     def __init__(self, *command: str) -> None:
         if not command:

src/ewoks/_requirements/pip.py

@@ -1,37 +1,34 @@
 import importlib.metadata
-import logging
 import sys
-from typing import Any
-from typing import Dict
 from typing import List
 from typing import Literal
 from typing import Optional
 
-from .utils import pip_freeze
-from .utils.base_manager import BaseManager
-from .utils.base_manager import BaseManagerInfo
-from .utils.base_manager import BaseRequirements
+from .utils.freeze_manager import FreezeManager
+from .utils.freeze_manager import FreezeManagerInfo
+from .utils.freeze_manager import FreezeRequirements
 
-logger = logging.getLogger(__name__)
 
-
-class PipManagerInfo(BaseManagerInfo):
+class PipManagerInfo(FreezeManagerInfo):
     name: Literal["pip"] = "pip"
-    freeze: List[str]
 
 
-class PipRequirements(BaseRequirements):
+class PipRequirements(FreezeRequirements):
     manager: PipManagerInfo
 
-    def __info__(self) -> str:
-        freeze = "\n  ".join(self.manager.freeze)
-        return f"{super().__info__()}\nRequirements:\n  {freeze}"
 
+class PipManager(FreezeManager):
+    """Uses the `pip` module of a python interpreter.
+
+    The command selects the target environment. For example
+    `PipManager("/path/to/python", "-m", "pip")`. The current python
+    environment is the target when no command is provided.
+    """
 
-class PipManager(BaseManager):
     NAME = "pip"
     PRIORITY = 0
     REQUIREMENTS_MODEL = PipRequirements
+    COMMAND_EXAMPLE = "python -m pip"
 
     def __init__(self, *command: str) -> None:
         if not command:
@@ -49,44 +46,8 @@ def is_active(self) -> bool:
         """Manager is explicitly active."""
         return False
 
-    def _gather_requirements(self) -> Dict[str, Any]:
-        freeze_output = self._check_output("freeze").strip().splitlines()
-
-        return {"freeze": freeze_output}
-
-    def _install_native_requirements(self, requirements: PipRequirements) -> bool:
-        freeze = requirements.manager.freeze
-
-        if not freeze:
-            return False
+    def _freeze(self) -> str:
+        return self._check_output("freeze")
 
-        arguments = self._arguments(freeze)
+    def _install(self, arguments: List[str]) -> None:
         self._check_call("install", "--no-cache-dir", *arguments)
-        return True
-
-    def _install_base_requirements(self, requirements: BaseRequirements) -> bool:
-        freeze = self._freeze_distributions(requirements)
-        if not freeze:
-            return False
-
-        arguments = self._arguments(freeze)
-        self._check_call("install", "--no-cache-dir", *arguments)
-        return True
-
-    def _freeze_distributions(self, requirements: BaseRequirements) -> List[str]:
-        """
-        Pip freeze argument from list of distributions.
-        """
-        freeze = []
-        for dist in requirements.distributions:
-            lines, warnings = pip_freeze.freeze_distribution(dist)
-            for warning in warnings:
-                logger.warning(warning)
-            freeze.extend(lines)
-        return freeze
-
-    def _arguments(self, freeze: List[str]) -> List[str]:
-        arguments, warnings = pip_freeze.sanitize_freeze(freeze)
-        for warning in warnings:
-            logger.warning(warning)
-        return arguments

src/ewoks/_requirements/pipenv.py

@@ -30,6 +30,7 @@ class PipenvManager(BaseManager):
     NAME = "pipenv"
     PRIORITY = 3
     REQUIREMENTS_MODEL = PipenvRequirements
+    COMMAND_EXAMPLE = "python -m pipenv"
 
     def __init__(self, *command: str) -> None:
         if not command:

src/ewoks/_requirements/pixi.py

@@ -22,6 +22,7 @@ class PixiManager(BaseManager):
     NAME = "pixi"
     PRIORITY = 5
     REQUIREMENTS_MODEL = PixiRequirements
+    COMMAND_EXAMPLE = "pixi"
 
     def __init__(self, *command: str) -> None:
         if not command:

src/ewoks/_requirements/poetry.py

@@ -25,6 +25,7 @@ class PoetryManager(BaseManager):
     NAME = "poetry"
     PRIORITY = 2
     REQUIREMENTS_MODEL = PoetryRequirements
+    COMMAND_EXAMPLE = "python -m poetry"
 
     def __init__(self, *command: str) -> None:
         if not command:

src/ewoks/_requirements/utils/_supported.py

@@ -3,20 +3,20 @@
 from typing import Type
 
 from ..pip import PipManager
+from ..uv import UvManager
 from .base_manager import BaseManager
 
 # from ..conda import CondaManager
 # from ..pipenv import PipenvManager
 # from ..pixi import PixiManager
 # from ..poetry import PoetryManager
-# from ..uv import UvManager
 
 
 @lru_cache(1)
 def get_supported_managers() -> Dict[str, Type[BaseManager]]:
     managers = [
         PipManager,
-        # UvManager,
+        UvManager,
         # PoetryManager,
         # PipenvManager,
         # CondaManager,

src/ewoks/_requirements/utils/base_manager.py

@@ -57,6 +57,7 @@ class BaseManager:
     NAME = NotImplemented
     PRIORITY = NotImplemented
     REQUIREMENTS_MODEL = NotImplemented
+    COMMAND_EXAMPLE = NotImplemented  # example of an associated shell command
 
     def __init__(self, *command: str) -> None:
         if not command:
@@ -114,7 +115,7 @@ def install_requirements(self, requirements: BaseRequirements) -> None:
     def _install_requirements(self, requirements: BaseRequirements) -> None:
         reraise = None
 
-        if isinstance(requirements, self.REQUIREMENTS_MODEL):
+        if self._is_native(requirements):
             try:
                 if self._install_native_requirements(requirements):
                     return
@@ -137,6 +138,10 @@ def _install_requirements(self, requirements: BaseRequirements) -> None:
             raise reraise
         raise ValueError("No distibutions provided to install")
 
+    def _is_native(self, requirements: BaseRequirements) -> bool:
+        """Requirements can be installed with the manager-specific method."""
+        return isinstance(requirements, self.REQUIREMENTS_MODEL)
+
     @abstractmethod
     def _gather_requirements(self) -> Dict[str, Any]:
         pass

src/ewoks/_requirements/utils/detect.py

@@ -44,7 +44,7 @@ def _select_manager(manager_name: str, manager_command: Tuple[str, ...]) -> Base
         raise ValueError(f"Package manager {manager_name!r} is not supported")
 
     manager = manager_cls(*manager_command)
-    if not manager.version:
+    if manager.version() is None:
         raise ValueError(f"Package manager {manager_name!r} is not available")
 
     return manager

src/ewoks/_requirements/utils/freeze_manager.py

@@ -0,0 +1,81 @@
+import logging
+from abc import abstractmethod
+from typing import Any
+from typing import Dict
+from typing import List
+
+from . import pip_freeze
+from .base_manager import BaseManager
+from .base_manager import BaseManagerInfo
+from .base_manager import BaseRequirements
+
+logger = logging.getLogger(__name__)
+
+
+class FreezeManagerInfo(BaseManagerInfo):
+    freeze: List[str]
+
+
+class FreezeRequirements(BaseRequirements):
+    manager: FreezeManagerInfo
+
+    def __info__(self) -> str:
+        freeze = "\n  ".join(self.manager.freeze)
+        return f"{super().__info__()}\nRequirements:\n  {freeze}"
+
+
+class FreezeManager(BaseManager):
+    """Interface for package managers that describe an environment with a
+    requirements list in `pip freeze` format.
+    """
+
+    def _is_native(self, requirements: BaseRequirements) -> bool:
+        """Any requirements list in `pip freeze` format can be installed,
+        also when generated by another package manager.
+        """
+        return isinstance(requirements, FreezeRequirements)
+
+    def _gather_requirements(self) -> Dict[str, Any]:
+        freeze = self._freeze().strip().splitlines()
+        return {"freeze": freeze}
+
+    def _install_native_requirements(self, requirements: FreezeRequirements) -> bool:
+        return self._install_freeze(requirements.manager.freeze)
+
+    def _install_base_requirements(self, requirements: BaseRequirements) -> bool:
+        return self._install_freeze(self._freeze_distributions(requirements))
+
+    @abstractmethod
+    def _freeze(self) -> str:
+        """Requirements of the current environment in `pip freeze` format."""
+        pass
+
+    @abstractmethod
+    def _install(self, arguments: List[str]) -> None:
+        """Install sanitized requirements in `pip freeze` format."""
+        pass
+
+    def _install_freeze(self, freeze: List[str]) -> bool:
+        if not freeze:
+            return False
+
+        self._install(self._arguments(freeze))
+        return True
+
+    def _freeze_distributions(self, requirements: BaseRequirements) -> List[str]:
+        """
+        Pip freeze argument from list of distributions.
+        """
+        freeze = []
+        for dist in requirements.distributions:
+            lines, warnings = pip_freeze.freeze_distribution(dist)
+            for warning in warnings:
+                logger.warning(warning)
+            freeze.extend(lines)
+        return freeze
+
+    def _arguments(self, freeze: List[str]) -> List[str]:
+        arguments, warnings = pip_freeze.sanitize_freeze(freeze)
+        for warning in warnings:
+            logger.warning(warning)
+        return arguments