--wip--

Author: tdstein

integration/tests/posit/connect/test_environments.py

@@ -0,0 +1,28 @@
+import select
+from posit import connect
+
+
+class TestContent:
+    @classmethod
+    def setup_class(cls):
+        cls.client = connect.Client()
+        cls.environment = cls.client.environments.create(
+            title="Title", cluster_name="Kubernetes", name="Name"
+        )
+
+    @classmethod
+    def teardown_class(cls):
+        cls.environment.destroy()
+
+    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):
+        self.environment.update(title="New Title")
+        assert self.environment["title"] == "New Title"

src/posit/connect/environments.py

@@ -1,3 +1,7 @@
+from typing import Dict, Literal, Optional, TypedDict, overload
+
+from typing_extensions import NotRequired, Required, Unpack
+
 from .resources import (
     ActiveCreatorMethods,
     ActiveDestroyerMethods,
@@ -7,12 +11,50 @@
 
 
 class Environment(ActiveUpdaterMethods, ActiveDestroyerMethods):
-    pass
+    class _UpdateEnvironment(TypedDict, total=False):
+        title: Required[str]
+        description: NotRequired[Optional[str]]
+        matching: NotRequired[Optional[Literal["any", "exact", "none"]]]
+        supervisor: NotRequired[Optional[str]]
+        python: NotRequired[Dict]
+        quarto: NotRequired[Dict]
+        r: NotRequired[Dict]
+        tensorflow: NotRequired[Dict]
+
+    @overload
+    def update(self, /, **attributes: Unpack[_UpdateEnvironment]): ...
+
+    @overload
+    def update(self, /, **attributes): ...
+
+    def update(self, /, **attributes):
+        return super().update(**attributes)
 
 
 class Environments(ActiveFinderMethods[Environment], ActiveCreatorMethods[Environment]):
     def __init__(self, ctx, path, pathinfo="environments", uid="guid"):
         super().__init__(ctx, path, pathinfo, uid)
 
+    class _CreateEnvironment(TypedDict, total=False):
+        title: Required[str]
+        description: NotRequired[Optional[str]]
+        cluster_name: Required[Literal["Kubernetes"]]
+        name: Required[str]
+        matching: NotRequired[Optional[Literal["any", "exact", "none"]]]
+        supervisor: NotRequired[Optional[str]]
+        python: NotRequired[Dict]
+        quarto: NotRequired[Dict]
+        r: NotRequired[Dict]
+        tensorflow: NotRequired[Dict]
+
+    @overload
+    def create(self, **attributes: Unpack[_CreateEnvironment]): ...
+
+    @overload
+    def create(self, **attributes): ...
+
+    def create(self, **attributes):
+        return super().create(**attributes)
+
     def _create_instance(self, path, pathinfo, /, **attributes):
         return Environment(self._ctx, path, pathinfo, **attributes)

src/posit/connect/resources.py

@@ -238,19 +238,15 @@ def create(self, **attributes) -> T:
         return self._create_instance(self._path, uid, **result)
 
 
-class ActiveDestroyerMethods(Active):
+class ActiveDestroyerMethods(Active, ABC):
     def destroy(self):
         endpoint = self._ctx.url + self._path
         self._ctx.session.delete(endpoint)
 
 
-class ActiveUpdaterMethods(Active):
+class ActiveUpdaterMethods(Active, ABC):
     def update(self, /, **attributes):
-        super().update(**attributes)
-        self.save()
-
-    def save(self):
         endpoint = self._ctx.url + self._path
-        response = self._ctx.session.put(endpoint, json=self)
+        response = self._ctx.session.put(endpoint, json=attributes)
         result = response.json()
         super().update(**result)