ewoks install: support pixi, conda, poetry, uv and pipenv

Author: woutdenolf

src/ewoks/__main__.py

@@ -159,7 +159,7 @@ def command_install(
     cli_install_utils.parse_install_arguments(cli_args, shell=shell)
     for workflow, graph in zip(cli_args.workflows, cli_args.graphs):
         try:
-            install_graph(graph, cli_args.yes, cli_args.python)
+            install_graph(graph, cli_args.yes, command=cli_args.manager_command)
         except CalledProcessError as e:
             print(f"Install failed for {workflow}: {e}")
         except AbortException:

src/ewoks/_requirements/__init__.py

@@ -1 +1,78 @@
 """Workflow requirements."""
+
+import logging
+from typing import Tuple
+
+from ewokscore.graph import TaskGraph
+
+from .managers.utils.base import BaseRequirements
+from .managers.utils.detect import get_manager
+from .metadata import parse
+from .metadata.gather import last_resort
+
+logger = logging.getLogger(__file__)
+
+
+def add_requirements(graph: TaskGraph, command: Tuple[str, ...] = tuple()) -> None:
+    """Add requirements to a workflow definition in-place."""
+    manager = get_manager(command=command)
+    requirements = manager.gather_requirements()
+    graph.graph.graph["requirements"] = requirements.model_dump()
+
+
+def get_requirements(graph: TaskGraph) -> BaseRequirements:
+    """Extract requirements from a workflow definition."""
+    requirements = graph.graph.graph.get("requirements", None)
+    no_requirements = not requirements
+
+    if no_requirements:
+        logger.warning(
+            "BaseRequirements field is empty. Trying to extract requirements automatically..."
+        )
+        requirements = last_resort.last_resort_requirements(graph)
+
+    requirements = parse.parse_requirements(requirements)
+
+    if no_requirements:
+        logger.info(f"Extracted the following requirements: {requirements.__info__()}")
+
+    return requirements
+
+
+def install_requirements(
+    requirements: BaseRequirements, command: Tuple[str, ...] = tuple()
+) -> None:
+    """Install workflow requirements."""
+    manager = get_manager(manager_name=requirements.manager.name, command=command)
+    manager.install_requirements(requirements)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.DEBUG)
+    import time
+    from pprint import pprint
+
+    t0 = time.perf_counter()
+
+    try:
+        manager = get_manager(manager_name=None)
+        requirements = manager.gather_requirements()
+
+        print()
+        print("Model:")
+        pprint(requirements.model_dump())
+    finally:
+        print("Freeze time:", time.perf_counter() - t0)
+
+    pip_freeze = requirements.manager.freeze
+
+    dists_freeze = manager.freeze_distributions(requirements)
+    dists_freeze = [s for s in dists_freeze if not s.startswith("#")]
+
+    print()
+    print("pip freeze has these extra's:")
+    pprint(set(pip_freeze) - set(dists_freeze))
+
+    print()
+    print("native freeze has these extra's:")
+    pprint(set(dists_freeze) - set(pip_freeze))

src/ewoks/_requirements/managers/__init__.py

@@ -0,0 +1,30 @@
+import logging
+
+import yaml
+
+from ..metadata.gather import gather_requirements
+from ..models.conda import CondaRequirements
+from .utils.base import BaseManager
+
+logger = logging.getLogger(__name__)
+
+
+class CondaManager(BaseManager):
+    NAME = "conda"
+
+    def _gather_requirements(self, manager_version: str) -> CondaRequirements:
+        output = self._check_output("env", "export")
+        environment = yaml.safe_load(output)
+        environment.pop("name", None)
+        environment.pop("prefix", None)
+
+        return gather_requirements(
+            manager_name="conda",
+            manager_version=manager_version,
+            environment=environment,
+        )
+
+    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)

src/ewoks/_requirements/managers/conda.py

@@ -0,0 +1,56 @@
+import logging
+from typing import List
+
+from ..metadata import pip_freeze
+from ..metadata.gather import gather_requirements
+from ..models.pip import PipRequirements
+from .utils.base import BaseManager
+
+logger = logging.getLogger(__name__)
+
+
+class PipManager(BaseManager):
+    NAME = "pip"
+
+    def _gather_requirements(self, manager_version: str) -> PipRequirements:
+        freeze_output = self._check_output("freeze").strip().splitlines()
+        return gather_requirements(
+            manager_name=self.NAME,
+            manager_version=manager_version,
+            freeze=freeze_output,
+        )
+
+    def _install_requirements(self, requirements: PipRequirements) -> None:
+        freeze = requirements.manager.freeze
+
+        if freeze:
+            arguments = self._arguments(freeze)
+            try:
+                self._check_call("install", "--no-cache-dir", *arguments)
+                return
+            except Exception:
+                if not requirements.distributions:
+                    raise
+
+        freeze = self.freeze_distributions(requirements)
+        if freeze:
+            arguments = self._arguments(freeze)
+            self._check_call("install", "--no-cache-dir", *arguments)
+            return
+
+        raise ValueError("No distibutions provided to install")
+
+    def freeze_distributions(self, requirements: PipRequirements) -> List[str]:
+        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/managers/pip.py

@@ -0,0 +1,36 @@
+import json
+
+from ..metadata.gather import gather_requirements
+from ..models.pipenv import PipenvRequirements
+from .utils.base import BaseManager
+
+
+class PipenvManager(BaseManager):
+    NAME = "pipenv"
+
+    def _gather_requirements(self, manager_version: str) -> PipenvRequirements:
+        output = self._check_output("lock", "--requirements")
+        requirements = output.strip().splitlines()
+
+        return gather_requirements(
+            manager_name=self.NAME,
+            manager_version=manager_version,
+            requirements=requirements,
+        )
+
+    def _install_requirements(self, requirements: PipenvRequirements) -> None:
+        lock_data = {
+            "_meta": {"hash": {"sha256": "dummy"}},  # minimal metadata
+            "default": {
+                pkg.split("==")[0]: {"version": pkg.split("==")[1]}
+                for pkg in requirements.requirements
+            },
+            "develop": {
+                pkg.split("==")[0]: {"version": pkg.split("==")[1]}
+                for pkg in getattr(requirements, "dev_requirements", [])
+            },
+        }
+        text = json.dumps(lock_data, indent=2)
+
+        with self._temporary_file(text, ".lock") as tmp_path:
+            self._check_call("sync", "--ignore-pipfile", "-f", tmp_path)

src/ewoks/_requirements/managers/pipenv.py

@@ -0,0 +1,29 @@
+import os
+
+from ..metadata.gather import gather_requirements
+from ..models.pixi import PixiRequirements
+from .utils.base import BaseManager
+
+
+class PixiManager(BaseManager):
+    NAME = "pixi"
+
+    def _gather_requirements(self, manager_version: str) -> PixiRequirements:
+        if os.path.exists("pixi.lock"):
+            with open("pixi.lock", "r", encoding="utf-8") as f:
+                lock_content = f.read()
+        elif os.path.exists("pixi.toml"):
+            with open("pixi.toml", "r", encoding="utf-8") as f:
+                lock_content = f.read()
+        else:
+            raise RuntimeError("No pixi.lock or pixi.toml file found")
+
+        return gather_requirements(
+            manager_name=self.NAME,
+            manager_version=manager_version,
+            lockfile=lock_content,
+        )
+
+    def _install_requirements(self, requirements: PixiRequirements) -> None:
+        with self._temporary_file(requirements.lockfile, ".lock") as tmp_path:
+            self._check_call("install", cwd=os.path.dirname(tmp_path))

src/ewoks/_requirements/managers/pixi.py

@@ -0,0 +1,22 @@
+from ..metadata.gather import gather_requirements
+from ..models.poetry import PoetryRequirements
+from .utils.base import BaseManager
+
+
+class PoetryManager(BaseManager):
+    NAME = "poetry"
+
+    def _gather_requirements(self, manager_version: str) -> PoetryRequirements:
+        output = self._check_output("export", "--without-hashes")
+        requirements = output.strip().splitlines()
+
+        return gather_requirements(
+            manager_name=self.NAME,
+            manager_version=manager_version,
+            requirements=requirements,
+        )
+
+    def _install_requirements(self, requirements: PoetryRequirements) -> None:
+        text = "\n".join(requirements.requirements)
+        with self._temporary_file(text, ".txt") as tmp_path:
+            self._check_call("add", "--lock", "--file", tmp_path)

src/ewoks/_requirements/managers/poetry.py

@@ -0,0 +1,114 @@
+import logging
+import os
+import subprocess
+import tempfile
+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__)
+
+
+class BaseManager:
+    """Defines the interface all package managers must implement.
+
+    If `MyManager` is an impementation of this interface then
+    to get the Ewoks workflow requirements like this:
+
+    .. code-block:: python
+
+        manager = MyManager()
+        requirements = manager.gather_requirements()
+
+    Ewoks workflow requirements can be installed like this:
+
+    .. code-block:: python
+
+        manager = MyManager()
+        install_requirements.install_requirements(requirements)
+    """
+
+    NAME = NotImplemented
+
+    def __init__(self, *command: str) -> None:
+        if not command:
+            command = get_manager_command(self.NAME)
+        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:
+            raise RuntimeError(f"{self.NAME!r} is not installed")
+
+        try:
+            return self._gather_requirements(manager_version)
+        except Exception as ex:
+            logger.error(
+                "%s: failed to generate requirements (%s)", type(self).__name__, ex
+            )
+            return None
+
+    def install_requirements(self, requirements: BaseRequirements) -> None:
+        """Install requirements into the current python environment."""
+        try:
+            return self._install_requirements(requirements)
+        except Exception as ex:
+            logger.error(
+                "%s: failed to install requirements (%s)", type(self).__name__, ex
+            )
+            raise
+
+    @abstractmethod
+    def _gather_requirements(self, manager_version: str) -> BaseRequirements:
+        pass
+
+    @abstractmethod
+    def _install_requirements(self, requirements: BaseRequirements) -> None:
+        pass
+
+    def _check_output(self, *args) -> str:
+        return _check_output([*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])
+
+    @contextmanager
+    def _temporary_file(self, text: str, suffix: str) -> Generator[str, None, None]:
+        tmp_path = None
+        try:
+            with tempfile.NamedTemporaryFile("w", suffix=suffix, delete=False) as tmp:
+                tmp.write(text)
+                tmp_path = tmp.name
+
+            yield tmp_path
+
+        finally:
+            if tmp_path:
+                try:
+                    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