In classes where ._ctx is available... use it instead of .params

Author: schloerke

src/posit/connect/content.py

@@ -293,8 +293,8 @@ def create_repository(
     def delete(self) -> None:
         """Delete the content item."""
         path = f"v1/content/{self['guid']}"
-        url = self.params.url + path
-        self.params.session.delete(url)
+        url = self._ctx.url + path
+        self._ctx.session.delete(url)
 
     def deploy(self) -> tasks.Task:
         """Deploy the content.
@@ -313,8 +313,8 @@ def deploy(self) -> tasks.Task:
         None
         """
         path = f"v1/content/{self['guid']}/deploy"
-        url = self.params.url + path
-        response = self.params.session.post(url, json={"bundle_id": None})
+        url = self._ctx.url + path
+        response = self._ctx.session.post(url, json={"bundle_id": None})
         result = response.json()
         ts = tasks.Tasks(self.params)
         return ts.get(result["task_id"])
@@ -369,8 +369,8 @@ def restart(self) -> None:
             self.environment_variables.create(key, unix_epoch_in_seconds)
             self.environment_variables.delete(key)
             # GET via the base Connect URL to force create a new worker thread.
-            url = posixpath.join(dirname(self.params.url), f"content/{self['guid']}")
-            self.params.session.get(url)
+            url = posixpath.join(dirname(self._ctx.url), f"content/{self['guid']}")
+            self._ctx.session.get(url)
             return None
         else:
             raise ValueError(
@@ -440,8 +440,8 @@ def update(
         -------
         None
         """
-        url = self.params.url + f"v1/content/{self['guid']}"
-        response = self.params.session.patch(url, json=attrs)
+        url = self._ctx.url + f"v1/content/{self['guid']}"
+        response = self._ctx.session.patch(url, json=attrs)
         super().update(**response.json())
 
     # Relationships
@@ -597,8 +597,8 @@ def create(
         ContentItem
         """
         path = "v1/content"
-        url = self.params.url + path
-        response = self.params.session.post(url, json=attrs)
+        url = self._ctx.url + path
+        response = self._ctx.session.post(url, json=attrs)
         return ContentItem(self._ctx, **response.json())
 
     @overload
@@ -685,8 +685,8 @@ def find(self, include: Optional[str | list[Any]] = None, **conditions) -> List[
             conditions["owner_guid"] = self.owner_guid
 
         path = "v1/content"
-        url = self.params.url + path
-        response = self.params.session.get(url, params=conditions)
+        url = self._ctx.url + path
+        response = self._ctx.session.get(url, params=conditions)
         return [
             ContentItem(
                 self._ctx,
@@ -858,6 +858,6 @@ def get(self, guid: str) -> ContentItem:
         ContentItem
         """
         path = f"v1/content/{guid}"
-        url = self.params.url + path
-        response = self.params.session.get(url)
+        url = self._ctx.url + path
+        response = self._ctx.session.get(url)
         return ContentItem(self._ctx, **response.json())

src/posit/connect/groups.py

@@ -43,8 +43,8 @@ def members(self) -> GroupMembers:
     def delete(self) -> None:
         """Delete the group."""
         path = f"v1/groups/{self['guid']}"
-        url = self.params.url + path
-        self.params.session.delete(url)
+        url = self._ctx.url + path
+        self._ctx.session.delete(url)
 
 
 class GroupMembers(Resources):
@@ -58,8 +58,8 @@ def find(self) -> list[User]:
         from .users import User
 
         path = f"v1/groups/{self._group_guid}/members"
-        url = self.params.url + path
-        paginator = Paginator(self.params.session, url)
+        url = self._ctx.url + path
+        paginator = Paginator(self._ctx.session, url)
         member_dicts = paginator.fetch_results()
 
         # For each member in the group
@@ -74,8 +74,8 @@ def count(self) -> int:
         int
         """
         path = f"v1/groups/{self._group_guid}/members"
-        url = self.params.url + path
-        response = self.params.session.get(url, params={"page_size": 1})
+        url = self._ctx.url + path
+        response = self._ctx.session.get(url, params={"page_size": 1})
         result = response.json()
         return result["total"]
 
@@ -123,8 +123,8 @@ def create(self, **kwargs) -> Group:
         Group
         """
         path = "v1/groups"
-        url = self.params.url + path
-        response = self.params.session.post(url, json=kwargs)
+        url = self._ctx.url + path
+        response = self._ctx.session.post(url, json=kwargs)
         return Group(self._ctx, **response.json())
 
     @overload
@@ -150,8 +150,8 @@ def find(self, **kwargs):
         List[Group]
         """
         path = "v1/groups"
-        url = self.params.url + path
-        paginator = Paginator(self.params.session, url, params=kwargs)
+        url = self._ctx.url + path
+        paginator = Paginator(self._ctx.session, url, params=kwargs)
         results = paginator.fetch_results()
         return [
             Group(
@@ -184,8 +184,8 @@ def find_one(self, **kwargs) -> Group | None:
         Group | None
         """
         path = "v1/groups"
-        url = self.params.url + path
-        paginator = Paginator(self.params.session, url, params=kwargs)
+        url = self._ctx.url + path
+        paginator = Paginator(self._ctx.session, url, params=kwargs)
         pages = paginator.fetch_pages()
         results = (result for page in pages for result in page.results)
         groups = (
@@ -208,8 +208,8 @@ def get(self, guid: str) -> Group:
         -------
         Group
         """
-        url = self.params.url + f"v1/groups/{guid}"
-        response = self.params.session.get(url)
+        url = self._ctx.url + f"v1/groups/{guid}"
+        response = self._ctx.session.get(url)
         return Group(
             self._ctx,
             **response.json(),
@@ -223,7 +223,7 @@ def count(self) -> int:
         int
         """
         path = "v1/groups"
-        url = self.params.url + path
-        response: requests.Response = self.params.session.get(url, params={"page_size": 1})
+        url = self._ctx.url + path
+        response: requests.Response = self._ctx.session.get(url, params={"page_size": 1})
         result: dict = response.json()
         return result["total"]

src/posit/connect/users.py

@@ -56,9 +56,9 @@ def lock(self, *, force: bool = False):
             raise RuntimeError(
                 "You cannot lock your own account. Set force=True to override this behavior.",
             )
-        url = self.params.url + f"v1/users/{self['guid']}/lock"
+        url = self._ctx.url + f"v1/users/{self['guid']}/lock"
         body = {"locked": True}
-        self.params.session.post(url, json=body)
+        self._ctx.session.post(url, json=body)
         super().update(locked=True)
 
     def unlock(self):
@@ -77,9 +77,9 @@ def unlock(self):
 
         >>> user.unlock()
         """
-        url = self.params.url + f"v1/users/{self['guid']}/lock"
+        url = self._ctx.url + f"v1/users/{self['guid']}/lock"
         body = {"locked": False}
-        self.params.session.post(url, json=body)
+        self._ctx.session.post(url, json=body)
         super().update(locked=False)
 
     class UpdateUser(TypedDict):
@@ -125,8 +125,8 @@ def update(
 
         >>> user.update(first_name="Jane", last_name="Smith")
         """
-        url = self.params.url + f"v1/users/{self['guid']}"
-        response = self.params.session.put(url, json=kwargs)
+        url = self._ctx.url + f"v1/users/{self['guid']}"
+        response = self._ctx.session.put(url, json=kwargs)
         super().update(**response.json())
 
     @property
@@ -259,8 +259,8 @@ def create(self, **attributes: Unpack[CreateUser]) -> User:
         ... )
         """
         # todo - use the 'context' module to inspect the 'authentication' object and route to POST (local) or PUT (remote).
-        url = self.params.url + "v1/users"
-        response = self.params.session.post(url, json=attributes)
+        url = self._ctx.url + "v1/users"
+        response = self._ctx.session.post(url, json=attributes)
         return User(self._ctx, **response.json())
 
     class FindUser(TypedDict):
@@ -302,8 +302,8 @@ def find(self, **conditions: Unpack[FindUser]) -> List[User]:
 
         >>> users = client.find(account_status="locked|licensed")
         """
-        url = self.params.url + "v1/users"
-        paginator = Paginator(self.params.session, url, params={**conditions})
+        url = self._ctx.url + "v1/users"
+        paginator = Paginator(self._ctx.session, url, params={**conditions})
         results = paginator.fetch_results()
         return [
             User(
@@ -345,8 +345,8 @@ def find_one(self, **conditions: Unpack[FindUser]) -> User | None:
 
         >>> user = client.find_one(account_status="locked|licensed")
         """
-        url = self.params.url + "v1/users"
-        paginator = Paginator(self.params.session, url, params={**conditions})
+        url = self._ctx.url + "v1/users"
+        paginator = Paginator(self._ctx.session, url, params={**conditions})
         pages = paginator.fetch_pages()
         results = (result for page in pages for result in page.results)
         users = (
@@ -375,8 +375,8 @@ def get(self, uid: str) -> User:
         --------
         >>> user = client.get("123e4567-e89b-12d3-a456-426614174000")
         """
-        url = self.params.url + f"v1/users/{uid}"
-        response = self.params.session.get(url)
+        url = self._ctx.url + f"v1/users/{uid}"
+        response = self._ctx.session.get(url)
         return User(
             self._ctx,
             **response.json(),
@@ -390,7 +390,7 @@ def count(self) -> int:
         -------
         int
         """
-        url = self.params.url + "v1/users"
-        response = self.params.session.get(url, params={"page_size": 1})
+        url = self._ctx.url + "v1/users"
+        response = self._ctx.session.get(url, params={"page_size": 1})
         result: dict = response.json()
         return result["total"]