use requirements class

Author: efiop

src/isolate/backends/common.py

@@ -9,6 +9,7 @@
 import threading
 import time
 from contextlib import contextmanager, suppress
+from dataclasses import dataclass, field
 from functools import lru_cache
 from pathlib import Path
 from types import ModuleType
@@ -220,37 +221,40 @@ def _normalize(text: str | bytes) -> bytes:
     return hashlib.sha256(inner_text).hexdigest()
 
 
-def normalize_requirement_layers(
-    requirements: list[str] | list[list[str]],
-) -> list[list[str]]:
-    """Return requirements as ordered layers."""
-    if not requirements:
-        return []
-    if all(isinstance(item, str) for item in requirements):
-        return [list(cast(List[str], requirements))]
-    if all(isinstance(item, list) for item in requirements):
-        layers: list[list[str]] = []
-        for layer in requirements:
-            if not all(isinstance(item, str) for item in layer):
-                raise TypeError("Requirements layers must contain strings.")
-            layers.append(list(layer))
-        return layers
-    raise TypeError(
-        "Requirements must be a list of strings or list of lists of strings."
-    )
-
-
-def requirement_key_fields(requirements: list[str] | list[list[str]]) -> list[str]:
-    """Return a list of fields to include in environment keys."""
-    if not requirements:
-        return []
-    if all(isinstance(item, str) for item in requirements):
-        return list(cast(List[str], requirements))
+@dataclass
+class Requirements:
+    layers: list[list[str]] = field(default_factory=list)
+
+    @classmethod
+    def from_raw(
+        cls, raw: Requirements | list[str] | list[list[str]] | None
+    ) -> Requirements:
+        if raw is None:
+            return cls()
+        if isinstance(raw, Requirements):
+            return raw
+        if isinstance(raw, list) and all(isinstance(item, str) for item in raw):
+            return cls([list(cast(List[str], raw))])
+        if isinstance(raw, list) and all(isinstance(item, list) for item in raw):
+            layers: list[list[str]] = []
+            for layer in raw:
+                if not all(isinstance(item, str) for item in layer):
+                    raise TypeError("Requirements layers must contain strings.")
+                layers.append(list(layer))
+            return cls(layers)
+        raise TypeError(
+            "Requirements must be a list of strings or list of lists of strings."
+        )
 
-    fields: list[str] = []
-    for index, layer in enumerate(normalize_requirement_layers(requirements)):
-        fields.extend(layer)
-    return fields
+    def keys(self) -> list[str]:
+        if not self.layers:
+            return []
+        if len(self.layers) == 1:
+            return list(self.layers[0])
+        return [
+            f"layer{index}:\n" + "\n".join(layer)
+            for index, layer in enumerate(self.layers)
+        ]
 
 
 def active_python() -> str:

src/isolate/backends/container.py

@@ -3,10 +3,10 @@
 import sys
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any, ClassVar, cast
+from typing import Any, ClassVar
 
 from isolate.backends import BaseEnvironment
-from isolate.backends.common import requirement_key_fields, sha256_digest_of
+from isolate.backends.common import Requirements, sha256_digest_of
 from isolate.backends.settings import DEFAULT_SETTINGS, IsolateSettings
 from isolate.connections import PythonIPC
 
@@ -17,9 +17,7 @@ class ContainerizedPythonEnvironment(BaseEnvironment[Path]):
 
     image: dict[str, Any] = field(default_factory=dict)
     python_version: str | None = None
-    requirements: list[str] | list[list[str]] = field(
-        default_factory=lambda: cast(list[str] | list[list[str]], [])
-    )
+    requirements: Requirements = field(default_factory=Requirements)
     tags: list[str] = field(default_factory=list)
     resolver: str | None = None
 
@@ -29,7 +27,11 @@ def from_config(
         config: dict[str, Any],
         settings: IsolateSettings = DEFAULT_SETTINGS,
     ) -> BaseEnvironment:
-        environment = cls(**config)
+        prepared = dict(config)
+        prepared["requirements"] = Requirements.from_raw(
+            config.get("requirements") or []
+        )
+        environment = cls(**prepared)
         environment.apply_settings(settings)
         if environment.resolver not in ("uv", None):
             raise ValueError(
@@ -48,7 +50,7 @@ def key(self) -> str:
         dockerfile_str = self.image.get("dockerfile_str", "")
         return sha256_digest_of(
             dockerfile_str,
-            *requirement_key_fields(self.requirements),
+            *self.requirements.keys(),
             *sorted(self.tags),
             *extras,
         )

src/isolate/backends/virtualenv.py

@@ -6,17 +6,16 @@
 from dataclasses import dataclass, field
 from functools import partial
 from pathlib import Path
-from typing import Any, ClassVar, cast
+from typing import Any, ClassVar
 
 from isolate.backends import BaseEnvironment, EnvironmentCreationError
 from isolate.backends.common import (
+    Requirements,
     active_python,
     get_executable,
     get_executable_path,
     logged_io,
-    normalize_requirement_layers,
     optional_import,
-    requirement_key_fields,
     sha256_digest_of,
 )
 from isolate.backends.settings import DEFAULT_SETTINGS, IsolateSettings
@@ -31,9 +30,7 @@
 class VirtualPythonEnvironment(BaseEnvironment[Path]):
     BACKEND_NAME: ClassVar[str] = "virtualenv"
 
-    requirements: list[str] | list[list[str]] = field(
-        default_factory=lambda: cast(list[str] | list[list[str]], [])
-    )
+    requirements: Requirements = field(default_factory=Requirements)
     constraints_file: os.PathLike | None = None
     python_version: str | None = None
     extra_index_urls: list[str] = field(default_factory=list)
@@ -46,7 +43,11 @@ def from_config(
         config: dict[str, Any],
         settings: IsolateSettings = DEFAULT_SETTINGS,
     ) -> BaseEnvironment:
-        environment = cls(**config)
+        prepared = dict(config)
+        prepared["requirements"] = Requirements.from_raw(
+            config.get("requirements") or []
+        )
+        environment = cls(**prepared)
         environment.apply_settings(settings)
         if environment.resolver not in ("uv", None):
             raise ValueError(
@@ -69,7 +70,7 @@ def key(self) -> str:
         active_python_version = self.python_version or active_python()
         return sha256_digest_of(
             active_python_version,
-            *requirement_key_fields(self.requirements),
+            *self.requirements.keys(),
             *constraints,
             *self.extra_index_urls,
             *sorted(self.tags),
@@ -187,7 +188,7 @@ def create(self, *, force: bool = False) -> Path:
                     f"Failed to create the environment at '{venv_path}': {exc}"
                 )
 
-            for layer in normalize_requirement_layers(self.requirements):
+            for layer in self.requirements.layers:
                 self._install_packages(venv_path, layer)
             completion_marker.touch()
 

src/isolate/server/server.py

@@ -25,7 +25,7 @@
     EnvironmentCreationError,
     IsolateSettings,
 )
-from isolate.backends.common import active_python
+from isolate.backends.common import Requirements, active_python
 from isolate.backends.local import LocalPythonEnvironment
 from isolate.backends.virtualenv import VirtualPythonEnvironment
 from isolate.connections.grpc import AgentError, LocalPythonGRPC
@@ -266,7 +266,7 @@ def _run_task(self, task: RunTask) -> Iterator[definitions.PartialRunResult]:
                 primary_environment, "python_version", active_python()
             )
             agent_environ = VirtualPythonEnvironment(
-                requirements=AGENT_REQUIREMENTS,
+                requirements=Requirements.from_raw(AGENT_REQUIREMENTS),
                 python_version=python_version,
             )
             agent_environ.apply_settings(run_settings)

tests/test_backends.py

@@ -9,7 +9,7 @@
 import isolate
 import pytest
 from isolate.backends import BaseEnvironment, EnvironmentCreationError
-from isolate.backends.common import get_executable, sha256_digest_of
+from isolate.backends.common import Requirements, get_executable, sha256_digest_of
 from isolate.backends.conda import CondaEnvironment
 from isolate.backends.local import LocalPythonEnvironment
 from isolate.backends.pyenv import PyenvEnvironment, _get_pyenv_executable
@@ -402,7 +402,7 @@ def test_layered_requirements_in_key(self, tmp_path):
                 "requirements": [["pyjokes==0.6.0"], ["pip==23.0.1"]],
             },
         )
-        assert base.key == layered.key
+        assert base.key != layered.key
 
     def test_layered_requirements_order_affects_key(self, tmp_path):
         first = self.get_environment(
@@ -438,7 +438,7 @@ def cli_run(args):
         )
 
         environment = VirtualPythonEnvironment(
-            requirements=[["pip==23.0.1"], ["pyjokes==0.6.0"]],
+            requirements=Requirements.from_raw([["pip==23.0.1"], ["pyjokes==0.6.0"]]),
         )
         environment.apply_settings(IsolateSettings(Path(tmp_path)))
         environment.create()
@@ -459,6 +459,28 @@ def test_try_using_uv(self, tmp_path):
         assert pyjokes_version == "0.5.0"
 
 
+class TestRequirements:
+    def test_from_raw_single_layer(self):
+        requirements = Requirements.from_raw(["a", "b"])
+        assert requirements.layers == [["a", "b"]]
+
+    def test_from_raw_multi_layer(self):
+        requirements = Requirements.from_raw([["a"], ["b", "c"]])
+        assert requirements.layers == [["a"], ["b", "c"]]
+
+    def test_from_raw_none(self):
+        requirements = Requirements.from_raw(None)
+        assert requirements.layers == []
+
+    def test_keys_single_layer_backward_compat(self):
+        requirements = Requirements.from_raw(["a", "b"])
+        assert requirements.keys() == ["a", "b"]
+
+    def test_keys_multi_layer_prefixes(self):
+        requirements = Requirements.from_raw([["a"], ["b", "c"]])
+        assert requirements.keys() == ["layer0:\na", "layer1:\nb\nc"]
+
+
 # Since mamba is an external dependency, we'll skip tests using it
 # if it is not installed.
 try:
@@ -933,7 +955,7 @@ def test_pyenv_environment(python_version, tmp_path):
 @pytest.mark.skipif(not IS_PYENV_AVAILABLE, reason="Pyenv is not available")
 def test_virtual_env_custom_python_version_with_pyenv(tmp_path, monkeypatch):
     pyjokes_env = VirtualPythonEnvironment(
-        requirements=["pyjokes==0.6.0"],
+        requirements=Requirements.from_raw(["pyjokes==0.6.0"]),
         python_version="3.9",
     )
 

tests/test_connections.py

@@ -7,6 +7,7 @@
 
 import pytest
 from isolate.backends import BaseEnvironment, EnvironmentConnection
+from isolate.backends.common import Requirements
 from isolate.backends.local import LocalPythonEnvironment
 from isolate.backends.settings import IsolateSettings
 from isolate.backends.virtualenv import VirtualPythonEnvironment
@@ -42,7 +43,7 @@ def make_venv(
         self, tmp_path: Any, requirements: List[str]
     ) -> VirtualPythonEnvironment:
         """Create a new virtual env with the specified requirements."""
-        env = VirtualPythonEnvironment(requirements)
+        env = VirtualPythonEnvironment(Requirements.from_raw(requirements))
         env.apply_settings(IsolateSettings(Path(tmp_path)))
         return env
 
@@ -206,6 +207,8 @@ def make_venv(
     ) -> VirtualPythonEnvironment:
         # Since gRPC agent requires isolate to be installed, we
         # have to add it to the requirements.
-        env = VirtualPythonEnvironment(requirements + [f"{REPO_DIR}[grpc]"])
+        env = VirtualPythonEnvironment(
+            Requirements.from_raw(requirements + [f"{REPO_DIR}[grpc]"])
+        )
         env.apply_settings(IsolateSettings(Path(tmp_path)))
         return env