--wip-- [skip ci]

Author: tdstein

src/posit/connect/client.py

@@ -6,6 +6,8 @@
 
 from requests import Response, Session
 
+from posit.connect.vanity import Vanities
+
 from . import hooks, me
 from .auth import Auth
 from .config import Config
@@ -271,6 +273,10 @@ def oauth(self) -> OAuth:
         """
         return OAuth(self.resource_params, self.cfg.api_key)
 
+    @property
+    def vanities(self) -> Vanities:
+        return Vanities(self.ctx)
+
     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/content.py

@@ -7,14 +7,14 @@
 from posixpath import dirname
 from typing import Any, List, Literal, Optional, overload
 
-from posit.connect.oauth.associations import ContentItemAssociations
-
 from . import tasks
 from .bundles import Bundles
 from .env import EnvVars
+from .oauth.associations import ContentItemAssociations
 from .permissions import Permissions
 from .resources import Resource, ResourceParameters, Resources
 from .tasks import Task
+from .vanity import VanityContentMixin
 from .variants import Variants
 
 

src/posit/connect/context.py

@@ -1,6 +1,7 @@
 import functools
 from typing import Optional, Protocol
 
+import requests
 from packaging.version import Version
 
 
@@ -21,7 +22,7 @@ def wrapper(instance: ContextManager, *args, **kwargs):
 
 
 class Context(dict):
-    def __init__(self, session, url):
+    def __init__(self, session: requests.Session, url: str):
         self.session = session
         self.url = url
 

src/posit/connect/vanity.py

@@ -0,0 +1,60 @@
+from typing import Callable, Optional
+
+from .context import Context
+
+AfterDestroyCallback = Callable[[], None]
+
+
+class Vanity(dict):
+    def __init__(
+        self, ctx: Context, *, after_destroy: AfterDestroyCallback = lambda: None, **kwargs
+    ):
+        super().__init__(**kwargs)
+        self._ctx = ctx
+        self._after_destroy = after_destroy
+
+    def destroy(self):
+        url = self._ctx.url + f"v1/content/{self['content_guid']}/vanity"
+        self._ctx.session.delete(url)
+        self._after_destroy()
+
+
+class Vanities:
+    def __init__(self, ctx: Context) -> None:
+        self._ctx = ctx
+
+    def all(self) -> list[Vanity]:
+        url = self._ctx.url + f"v1/vanities"
+        response = self._ctx.session.get(url)
+        results = response.json()
+        return [Vanity(self._ctx, **result) for result in results]
+
+
+class VanityContentMixin(dict):
+    def __init__(self, ctx: Context, **kwargs):
+        super().__init__(**kwargs)
+        self._ctx = ctx
+        self._vanity: Optional[Vanity] = None
+
+    @property
+    def vanity(self) -> Vanity:
+        if self._vanity is None:
+            url = self._ctx.url + f"v1/content/{self['guid']}/vanity"
+            response = self._ctx.session.get(url)
+            vanity_data = response.json()
+            # Set the after_destroy callback to reset _vanity to None when destroyed
+            after_destroy = lambda: setattr(self, "_vanity", None)
+            self._vanity = Vanity(self._ctx, after_destroy=after_destroy, **vanity_data)
+        return self._vanity
+
+    @vanity.setter
+    def vanity(self, value: dict):
+        url = self._ctx.url + f"v1/content/{self['guid']}/vanity"
+        self._ctx.session.put(url, json=value)
+        # Refresh the vanity property to reflect the updated value
+        self._vanity = self.vanity
+
+    @vanity.deleter
+    def vanity(self):
+        if self._vanity:
+            self._vanity.destroy()