feat: switch User and Content to inherit dict, add update method for User

src/posit/connect/content.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Callable, List, TypedDict
+from typing import Any, Callable, List, Optional
 
 from requests import Session
 
@@ -10,9 +10,181 @@
 from .resources import Resources
 
 
-class ContentItem(TypedDict, total=False):
-    # TODO: specify types
-    pass
+class ContentItem(dict):
+    @property
+    def guid(self) -> str:
+        return self.get("guid")  # type: ignore
+
+    @property
+    def name(self) -> str:
+        return self.get("name")  # type: ignore
+
+    @property
+    def title(self) -> Optional[str]:
+        return self.get("title")  # type: ignore
+
+    @property
+    def description(self) -> str:
+        return self.get("description")  # type: ignore
+
+    @property
+    def access_type(self) -> str:
+        return self.get("access_type")  # type: ignore
+
+    @property
+    def connection_timeout(self) -> Optional[int]:
+        return self.get("connection_timeout")  # type: ignore
+
+    @property
+    def read_timeout(self) -> Optional[int]:
+        return self.get("read_timeout")  # type: ignore
+
+    @property
+    def init_timeout(self) -> Optional[int]:
+        return self.get("init_timeout")  # type: ignore
+
+    @property
+    def idle_timeout(self) -> Optional[int]:
+        return self.get("idle_timeout")  # type: ignore
+
+    @property
+    def max_processes(self) -> Optional[int]:
+        return self.get("max_processes")  # type: ignore
+
+    @property
+    def min_processes(self) -> Optional[int]:
+        return self.get("min_processes")  # type: ignore
+
+    @property
+    def max_conns_per_process(self) -> Optional[int]:
+        return self.get("max_conns_per_process")  # type: ignore
+
+    @property
+    def load_factor(self) -> Optional[float]:
+        return self.get("load_factor")  # type: ignore
+
+    @property
+    def cpu_request(self) -> Optional[float]:
+        return self.get("cpu_request")  # type: ignore
+
+    @property
+    def cpu_limit(self) -> Optional[float]:
+        return self.get("cpu_limit")  # type: ignore
+
+    @property
+    def memory_request(self) -> Optional[int]:
+        return self.get("memory_request")  # type: ignore
+
+    @property
+    def memory_limit(self) -> Optional[int]:
+        return self.get("memory_limit")  # type: ignore
+
+    @property
+    def amd_gpu_limit(self) -> Optional[int]:
+        return self.get("amd_gpu_limit")  # type: ignore
+
+    @property
+    def nvidia_gpu_limit(self) -> Optional[int]:
+        return self.get("nvidia_gpu_limit")  # type: ignore
+
+    @property
+    def created_time(self) -> str:
+        return self.get("created_time")  # type: ignore
+
+    @property
+    def last_deployed_time(self) -> str:
+        return self.get("last_deployed_time")  # type: ignore
+
+    @property
+    def bundle_id(self) -> Optional[str]:
+        return self.get("bundle_id")  # type: ignore
+
+    @property
+    def app_mode(self) -> str:
+        return self.get("app_mode")  # type: ignore
+
+    @property
+    def content_category(self) -> Optional[str]:
+        return self.get("content_category")  # type: ignore
+
+    @property
+    def parameterized(self) -> bool:
+        return self.get("parameterized")  # type: ignore
+
+    @property
+    def cluster_name(self) -> Optional[str]:
+        return self.get("cluster_name")  # type: ignore
+
+    @property
+    def image_name(self) -> Optional[str]:
+        return self.get("image_name")  # type: ignore
+
+    @property
+    def default_image_name(self) -> Optional[str]:
+        return self.get("default_image_name")  # type: ignore
+
+    @property
+    def default_r_environment_management(self) -> Optional[bool]:
+        return self.get("default_r_environment_management")  # type: ignore
+
+    @property
+    def default_py_environment_management(self) -> Optional[bool]:
+        return self.get("default_py_environment_management")  # type: ignore
+
+    @property
+    def service_account_name(self) -> Optional[str]:
+        return self.get("service_account_name")  # type: ignore
+
+    @property
+    def r_version(self) -> Optional[str]:
+        return self.get("r_version")  # type: ignore
+
+    @property
+    def r_environment_management(self) -> Optional[bool]:
+        return self.get("r_environment_management")  # type: ignore
+
+    @property
+    def py_version(self) -> Optional[str]:
+        return self.get("py_version")  # type: ignore
+
+    @property
+    def py_environment_management(self) -> Optional[bool]:
+        return self.get("py_environment_management")  # type: ignore
+
+    @property
+    def quarto_version(self) -> Optional[str]:
+        return self.get("quarto_version")  # type: ignore
+
+    @property
+    def run_as(self) -> Optional[str]:
+        return self.get("run_as")  # type: ignore
+
+    @property
+    def run_as_current_user(self) -> bool:
+        return self.get("run_as_current_user")  # type: ignore
+
+    @property
+    def owner_guid(self) -> str:
+        return self.get("owner_guid")  # type: ignore
+
+    @property
+    def content_url(self) -> str:
+        return self.get("content_url")  # type: ignore
+
+    @property
+    def dashboard_url(self) -> str:
+        return self.get("dashboard_url")  # type: ignore
+
+    @property
+    def app_role(self) -> str:
+        return self.get("app_role")  # type: ignore
+
+    @property
+    def id(self) -> str:
+        return self.get("id")  # type: ignore
+
+    def __setattr__(self, name: str, value: Any) -> None:
+        raise AttributeError("Cannot set attributes: use update() instead")
 
 
 class Content(Resources[ContentItem]):

src/posit/connect/users.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
-
-from datetime import datetime
-from typing import Callable, List, TypedDict
+from typing import Any, Callable, List, Optional
 
 from requests import Session
 
@@ -13,18 +11,90 @@
 from .resources import Resources
 
 
-class User(TypedDict, total=False):
-    guid: str
-    email: str
-    username: str
-    first_name: str
-    last_name: str
-    user_role: str
-    created_time: datetime
-    updated_time: datetime
-    active_time: datetime
-    confirmed: bool
-    locked: bool
+class User(dict):
+
+    @property
+    def guid(self) -> str:
+        return self.get("guid")  # type: ignore
+
+    @property
+    def email(self) -> str:
+        return self.get("email")  # type: ignore
+
+    @property
+    def username(self) -> str:
+        return self.get("username")  # type: ignore
+
+    @property
+    def first_name(self) -> str:
+        return self.get("first_name")  # type: ignore
+
+    @property
+    def last_name(self) -> str:
+        return self.get("last_name")  # type: ignore
+
+    @property
+    def user_role(self) -> str:
+        return self.get("user_role")  # type: ignore
+
+    @property
+    def created_time(self) -> str:
+        return self.get("created_time")  # type: ignore
+
+    @property
+    def updated_time(self) -> str:
+        return self.get("updated_time")  # type: ignore
+
+    @property
+    def active_time(self) -> str:
+        return self.get("active_time")  # type: ignore
+
+    @property
+    def confirmed(self) -> bool:
+        return self.get("confirmed")  # type: ignore
+
+    @property
+    def locked(self) -> bool:
+        return self.get("locked")  # type: ignore
+
+    def __setattr__(self, name: str, value: Any) -> None:
+        raise AttributeError("Cannot set attributes: use update() instead.")
+
+    def _update(self, body):
+        self.get("session").patch(self.get("url"), json=body)
+        # If the request is successful, update the local object
+        super().update(body)
+        # TODO(#99): that patch request returns a payload on success,
+        # so we should instead update the local object with that payload
+        # (includes updated_time)
+
+    def update(  # type: ignore
+        self,
+        # Not all properties are settable, so we enumerate them here
+        # (also for type-hinting purposes)
+        email: Optional[str] = None,
+        username: Optional[str] = None,
+        first_name: Optional[str] = None,
+        last_name: Optional[str] = None,
+        user_role: Optional[str] = None,
+        # TODO(#100): in the API, this goes via POST /v1/users/{guid}/lock
+        # accept it here and make that request? Or add a .lock() method?
+        # locked: Optional[bool] = None,
+    ) -> None:
+        kwargs = {}
+        if email is not None:
+            kwargs["email"] = email
+        if username is not None:
+            kwargs["username"] = username
+        if first_name is not None:
+            kwargs["first_name"] = first_name
+        if last_name is not None:
+            kwargs["last_name"] = last_name
+        if user_role is not None:
+            kwargs["user_role"] = user_role
+        # if locked is not None:
+        #     kwargs["locked"] = locked
+        self._update(kwargs)
 
 
 class Users(Resources[User]):
@@ -37,7 +107,15 @@ def find(
         self, filter: Callable[[User], bool] = lambda _: True, page_size=_MAX_PAGE_SIZE
     ) -> List[User]:
         results = Paginator(self.session, self.url, page_size=page_size).get_all()
-        return [User(**user) for user in results if filter(User(**user))]
+        return [
+            User(
+                **user,
+                session=self.session,
+                url=urls.append_path(self.url, user["guid"]),
+            )
+            for user in results
+            if filter(User(**user))
+        ]
 
     def find_one(
         self, filter: Callable[[User], bool] = lambda _: True, page_size=_MAX_PAGE_SIZE
@@ -46,15 +124,24 @@ def find_one(
         while pager.total is None or pager.seen < pager.total:
             result = pager.get_next_page()
             for u in result:
-                user = User(**u)
+                user = User(
+                    **u,
+                    session=self.session,
+                    url=urls.append_path(self.url, u["guid"]),
+                )
                 if filter(user):
                     return user
         return None
 
     def get(self, id: str) -> User:
         url = urls.append_path(self.url, id)
         response = self.session.get(url)
-        return User(**response.json())
+        raw_user = response.json()
+        return User(
+            **raw_user,
+            session=self.session,
+            url=urls.append_path(self.url, raw_user["guid"]),
+        )
 
     def create(self) -> User:
         raise NotImplementedError()

tests/posit/connect/test_content.py

@@ -1,6 +1,7 @@
 import responses
 
 from posit.connect.client import Client
+from posit.connect.content import ContentItem
 
 from .api import load_mock  # type: ignore
 
@@ -15,7 +16,7 @@ def test_get_all_content(self):
         con = Client("12345", "https://connect.example")
         all_content = con.content.find()
         assert len(all_content) == 3
-        assert [c["name"] for c in all_content] == [
+        assert [c.name for c in all_content] == [
             "team-admin-dashboard",
             "Performance-Data-1671216053560",
             "My-Streamlit-app",
@@ -29,11 +30,12 @@ def test_content_find_one(self):
         )
         con = Client("12345", "https://connect.example")
 
-        one = con.content.find_one(lambda c: c["title"] == "Performance Data")
-        assert one["name"] == "Performance-Data-1671216053560"
+        one = con.content.find_one(lambda c: c.title == "Performance Data")
+        assert isinstance(one, ContentItem)
+        assert one.name == "Performance-Data-1671216053560"
 
         # Test find_one doesn't find any
-        assert con.content.find_one(lambda c: c["title"] == "Does not exist") is None
+        assert con.content.find_one(lambda c: c.title == "Does not exist") is None
 
     @responses.activate
     def test_content_get(self):
@@ -43,4 +45,4 @@ def test_content_get(self):
         )
         con = Client("12345", "https://connect.example")
         get_one = con.content.get("f2f37341-e21d-3d80-c698-a935ad614066")
-        assert get_one["name"] == "Performance-Data-1671216053560"
+        assert get_one.name == "Performance-Data-1671216053560"