feat: add environments

.coveragerc

@@ -1,8 +1,9 @@
 # This file contains the configuration settings for the coverage report generated.
 
 [report]
-# exclude '...' (ellipsis literal). This option uses regex, so an escaped literal is required.
 exclude_also =
     \.\.\.
+exclude_lines =
+    if TYPE_CHECKING:
 
 fail_under = 80

integration/tests/posit/connect/test_environments.py

@@ -0,0 +1,38 @@
+import pytest
+from packaging import version
+
+from posit import connect
+
+from . import CONNECT_VERSION
+
+
+@pytest.mark.skipif(
+    CONNECT_VERSION < version.parse("2023.05.0"),
+    reason="Environments API unavailable",
+)
+class TestEnvironments:
+    @classmethod
+    def setup_class(cls):
+        cls.client = connect.Client()
+        cls.environment = cls.client.environments.create(
+            title="title", name="name", cluster_name="Kubernetes"
+        )
+
+    @classmethod
+    def teardown_class(cls):
+        cls.environment.destroy()
+        assert len(cls.client.environments) == 0
+
+    def test_find(self):
+        uid = self.environment["guid"]
+        environment = self.client.environments.find(uid)
+        assert environment == self.environment
+
+    def test_find_by(self):
+        environment = self.client.environments.find_by(name="name")
+        assert environment == self.environment
+
+    def test_update(self):
+        assert self.environment["title"] == "title"
+        self.environment.update(title="new-title")
+        assert self.environment["title"] == "new-title"

src/posit/connect/client.py

@@ -2,12 +2,10 @@
 
 from __future__ import annotations
 
-from typing import overload
+from typing import TYPE_CHECKING, overload
 
 from requests import Response, Session
 
-from posit.connect.tags import Tags
-
 from . import hooks, me
 from .auth import Auth
 from .config import Config
@@ -17,11 +15,15 @@
 from .metrics import Metrics
 from .oauth import OAuth
 from .packages import Packages
-from .resources import ResourceParameters
+from .resources import ResourceParameters, _ResourceSequence
+from .tags import Tags
 from .tasks import Tasks
 from .users import User, Users
 from .vanities import Vanities
 
+if TYPE_CHECKING:
+    from .environments import Environments
+
 
 class Client(ContextManager):
     """
@@ -303,6 +305,11 @@ def packages(self) -> Packages:
     def vanities(self) -> Vanities:
         return Vanities(self.resource_params)
 
+    @property
+    @requires(version="2023.05.0")
+    def environments(self) -> Environments:
+        return _ResourceSequence(self._ctx, "v1/environments")
+
     def __del__(self):
         """Close the session when the Client instance is deleted."""
         if hasattr(self, "session") and self.session is not None:

src/posit/connect/environments.py

@@ -0,0 +1,211 @@
+from __future__ import annotations
+
+from abc import abstractmethod
+from collections.abc import Mapping, Sized
+from typing import (
+    Any,
+    List,
+    Literal,
+    Protocol,
+    TypedDict,
+    runtime_checkable,
+)
+
+MatchingType = Literal["any", "exact", "none"]
+"""Directions for how environments are considered for selection.
+
+    - any: The image may be selected by Connect if not defined in the bundle manifest.
+    - exact: The image must be defined in the bundle manifest
+    - none: Never use this environment
+"""
+
+
+class Installation(TypedDict):
+    """Interpreter installation in an execution environment."""
+
+    path: str
+    """The absolute path to the interpreter's executable."""
+
+    version: str
+    """The semantic version of the interpreter."""
+
+
+class Installations(TypedDict):
+    """Interpreter installations in an execution environment."""
+
+    installations: List[Installation]
+    """Interpreter installations in an execution environment."""
+
+
+class Environment(Mapping[str, Any]):
+    @abstractmethod
+    def destroy(self) -> None:
+        """Destroy the environment.
+
+        Warnings
+        --------
+        This operation is irreversible.
+
+        Note
+        ----
+        This action requires administrator privileges.
+        """
+
+    @abstractmethod
+    def update(
+        self,
+        *,
+        title: str,
+        description: str | None = ...,
+        matching: MatchingType | None = ...,
+        supervisor: str | None = ...,
+        python: Installations | None = ...,
+        quarto: Installations | None = ...,
+        r: Installations | None = ...,
+        tensorflow: Installations | None = ...,
+    ) -> None:
+        """Update the environment.
+
+        Parameters
+        ----------
+        title : str
+            A human-readable title.
+        description : str | None, optional, not required
+            A human-readable description.
+        matching : MatchingType, optional, not required
+            Directions for how the environment is considered for selection
+        supervisor : str | None, optional, not required
+            Path to the supervisor script.
+        python : Installations, optional, not required
+            The Python installations available in this environment
+        quarto : Installations, optional, not required
+            The Quarto installations available in this environment
+        r : Installations, optional, not required
+            The R installations available in this environment
+        tensorflow : Installations, optional, not required
+            The Tensorflow installations available in this environment
+
+        Note
+        ----
+        This action requires administrator privileges.
+        """
+
+
+@runtime_checkable
+class Environments(Sized, Protocol):
+    def create(
+        self,
+        *,
+        title: str,
+        name: str,
+        cluster_name: str | Literal["Kubernetes"],
+        matching: MatchingType = "any",
+        description: str | None = ...,
+        supervisor: str | None = ...,
+        python: Installations | None = ...,
+        quarto: Installations | None = ...,
+        r: Installations | None = ...,
+        tensorflow: Installations | None = ...,
+    ) -> Environment:
+        """Create an environment.
+
+        Parameters
+        ----------
+        title : str
+            A human-readable title.
+        name : str
+            The container image name used for execution in this environment.
+        cluster_name : str | Literal["Kubernetes"]
+            The cluster identifier for this environment. Defaults to "Kubernetes" when Off-Host-Execution is enabled.
+        description : str, optional
+            A human-readable description.
+        matching : MatchingType
+            Directions for how the environment is considered for selection, by default is "any".
+        supervisor : str, optional
+            Path to the supervisor script
+        python : Installations, optional
+            The Python installations available in this environment
+        quarto : Installations, optional
+            The Quarto installations available in this environment
+        r : Installations, optional
+            The R installations available in this environment
+        tensorflow : Installations, optional
+            The Tensorflow installations available in this environment
+
+        Returns
+        -------
+        Environment
+
+        Note
+        ----
+        This action requires administrator privileges.
+        """
+        ...
+
+    def find(self, guid: str, /) -> Environment: ...
+
+    def find_by(
+        self,
+        *,
+        id: str = ...,  # noqa: A002
+        guid: str = ...,
+        created_time: str = ...,
+        updated_time: str = ...,
+        title: str = ...,
+        name: str = ...,
+        description: str | None = ...,
+        cluster_name: str | Literal["Kubernetes"] = ...,
+        environment_type: str | Literal["Kubernetes"] = ...,
+        matching: MatchingType = ...,
+        supervisor: str | None = ...,
+        python: Installations | None = ...,
+        quarto: Installations | None = ...,
+        r: Installations | None = ...,
+        tensorflow: Installations | None = ...,
+    ) -> Environment | None:
+        """Find the first record matching the specified conditions.
+
+        There is no implied ordering, so if order matters, you should specify it yourself.
+
+        Parameters
+        ----------
+        id : str
+            The numerical identifier.
+        guid : str
+            The unique identifier.
+        created_time : str
+            The timestamp (RFC3339) when the environment was created.
+        updated_time : str
+            The timestamp (RFC3339) when the environment was updated.
+        title : str
+            A human-readable title.
+        name : str
+            The container image name used for execution in this environment.
+        description : str, optional
+            A human-readable description.
+        cluster_name : str | Literal["Kubernetes"]
+            The cluster identifier for this environment. Defaults to "Kubernetes" when Off-Host-Execution is enabled.
+        environment_type : str | Literal["Kubernetes"]
+            The cluster environment type. Defaults to "Kubernetes" when Off-Host-Execution is enabled.
+        matching : MatchingType
+            Directions for how the environment is considered for selection.
+        supervisor : str, optional
+            Path to the supervisor script
+        python : Installations, optional
+            The Python installations available in this environment
+        quarto : Installations, optional
+            The Quarto installations available in this environment
+        r : Installations, optional
+            The R installations available in this environment
+        tensorflow : Installations, optional
+            The Tensorflow installations available in this environment
+
+        Returns
+        -------
+        Environment | None
+
+        Note
+        ----
+        This action requires administrator or publisher privileges.
+        """
+        ...

src/posit/connect/resources.py

@@ -234,3 +234,83 @@ def find_by(self, **conditions: Any) -> T | None:
         """
         collection = self.fetch(**conditions)
         return next((v for v in collection if v.items() >= conditions.items()), None)
+
+
+class _Resource(dict):
+    def __init__(self, ctx: Context, path: str, **attributes):
+        self._ctx = ctx
+        self._path = path
+        super().__init__(**attributes)
+
+    def destroy(self) -> None:
+        self._ctx.client.delete(self._path)
+
+    def update(self, **attributes):  # type: ignore[reportIncompatibleMethodOverride]
+        response = self._ctx.client.put(self._path, json=attributes)
+        result = response.json()
+        super().update(**result)
+
+
+class _ResourceSequence(Sequence):
+    def __init__(self, ctx: Context, path: str, *, uid: str = "guid"):
+        self._ctx = ctx
+        self._path = path
+        self._uid = uid
+
+    def __getitem__(self, index):
+        return self.fetch()[index]
+
+    def __len__(self) -> int:
+        return len(self.fetch())
+
+    def __iter__(self):
+        return iter(self.fetch())
+
+    def __str__(self) -> str:
+        return str(self.fetch())
+
+    def __repr__(self) -> str:
+        return repr(self.fetch())
+
+    def create(self, **attributes: Any) -> Any:
+        response = self._ctx.client.post(self._path, json=attributes)
+        result = response.json()
+        uid = result[self._uid]
+        path = posixpath.join(self._path, uid)
+        return _Resource(self._ctx, path, **result)
+
+    def fetch(self, **conditions) -> List[Any]:
+        response = self._ctx.client.get(self._path, params=conditions)
+        results = response.json()
+        resources = []
+        for result in results:
+            uid = result[self._uid]
+            path = posixpath.join(self._path, uid)
+            resource = _Resource(self._ctx, path, **result)
+            resources.append(resource)
+
+        return resources
+
+    def find(self, *args: str) -> Any:
+        path = posixpath.join(self._path, *args)
+        response = self._ctx.client.get(path)
+        result = response.json()
+        return _Resource(self._ctx, path, **result)
+
+    def find_by(self, **conditions) -> Any | None:
+        """
+        Find the first record matching the specified conditions.
+
+        There is no implied ordering, so if order matters, you should specify it yourself.
+
+        Parameters
+        ----------
+        **conditions : Any
+
+        Returns
+        -------
+        Optional[T]
+            The first record matching the conditions, or `None` if no match is found.
+        """
+        collection = self.fetch(**conditions)
+        return next((v for v in collection if v.items() >= conditions.items()), None)