refactor: introduce the active pattern

Author: tdstein

src/posit/connect/content.py

@@ -9,6 +9,7 @@
 
 from . import tasks
 from .bundles import Bundles
+from .context import Context
 from .env import EnvVars
 from .jobs import JobsMixin
 from .oauth.associations import ContentItemAssociations
@@ -34,6 +35,10 @@ class ContentItemOwner(Resource):
 
 
 class ContentItem(JobsMixin, VanityMixin, Resource):
+    def __init__(self, /, params: ResourceParameters, **kwargs):
+        ctx = Context(params.session, params.url)
+        super().__init__(ctx, **kwargs)
+
     def __getitem__(self, key: Any) -> Any:
         v = super().__getitem__(key)
         if key == "owner" and isinstance(v, dict):

src/posit/connect/context.py

@@ -40,7 +40,7 @@ def version(self) -> Optional[str]:
         return value
 
     @version.setter
-    def version(self, value: str):
+    def version(self, value):
         self["version"] = value
 
 

src/posit/connect/jobs.py

@@ -1,9 +1,8 @@
-from typing import List, Literal, Optional, Sequence, TypedDict, overload
+from typing import Literal, Optional, TypedDict, overload
 
 from typing_extensions import NotRequired, Required, Unpack
 
-from .errors import ClientError
-from .resources import FinderMethods, Resource, ResourceParameters, Resources
+from .resources import Active, ActiveFinderMethods, Resource
 
 JobTag = Literal[
     "unknown",
@@ -32,7 +31,7 @@
 ]
 
 
-class Job(Resource):
+class Job(Active):
     class _Job(TypedDict):
         # Identifiers
         id: Required[str]
@@ -100,10 +99,12 @@ class _Job(TypedDict):
         tag: Required[JobTag]
         """A tag categorizing the job type. Options are build_jupyter, build_report, build_site, configure_report, git, packrat_restore, python_restore, render_shiny, run_api, run_app, run_bokeh_app, run_dash_app, run_fastapi_app, run_pyshiny_app, run_python_api, run_streamlit, run_tensorflow, run_voila_app, testing, unknown, val_py_ext_pkg, val_r_ext_pkg, and val_r_install."""
 
-    def __init__(self, /, params, endpoint, **kwargs: Unpack[_Job]):
+    def __init__(self, /, params, **kwargs: Unpack[_Job]):
         super().__init__(params, **kwargs)
-        key = kwargs["key"]
-        self._endpoint = endpoint + key
+
+    @property
+    def _endpoint(self) -> str:
+        return self._ctx.url + f"v1/content/{self['app_id']}/jobs/{self['key']}"
 
     def destroy(self) -> None:
         """Destroy the job.
@@ -118,48 +119,21 @@ def destroy(self) -> None:
         ----
         This action requires administrator, owner, or collaborator privileges.
         """
-        self.params.session.delete(self._endpoint)
+        self._ctx.session.delete(self._endpoint)
 
 
-class Jobs(FinderMethods[Job], Sequence[Job], Resources):
+class Jobs(ActiveFinderMethods[Job]):
     """A collection of jobs."""
 
-    def __init__(self, params, endpoint):
-        super().__init__(Job, params, endpoint)
-        self._endpoint = endpoint + "jobs"
-        self._cache = None
-
-    @property
-    def _data(self) -> List[Job]:
-        if self._cache:
-            return self._cache
-
-        response = self.params.session.get(self._endpoint)
-        results = response.json()
-        self._cache = [Job(self.params, self._endpoint, **result) for result in results]
-        return self._cache
-
-    def __getitem__(self, index):
-        """Retrieve an item or slice from the sequence."""
-        return self._data[index]
-
-    def __len__(self):
-        """Return the length of the sequence."""
-        return len(self._data)
+    _uid = "key"
 
-    def __repr__(self):
-        """Return the string representation of the sequence."""
-        return f"Jobs({', '.join(map(str, self._data))})"
+    def __init__(self, cls, ctx, parent: Active):
+        super().__init__(cls, ctx)
+        self._parent = parent
 
-    def count(self, value):
-        """Return the number of occurrences of a value in the sequence."""
-        return self._data.count(value)
-
-    def index(self, value, start=0, stop=None):
-        """Return the index of the first occurrence of a value in the sequence."""
-        if stop is None:
-            stop = len(self._data)
-        return self._data.index(value, start, stop)
+    @property
+    def _endpoint(self) -> str:
+        return self._ctx.url + f"v1/content/{self._parent['guid']}/jobs"
 
     class _FindByRequest(TypedDict, total=False):
         # Identifiers
@@ -286,37 +260,18 @@ def find_by(self, **conditions: Unpack[_FindByRequest]) -> Optional[Job]:
     @overload
     def find_by(self, **conditions): ...
 
-    def find_by(self, **conditions):
-        if "key" in conditions and self._cache is None:
-            key = conditions["key"]
-            try:
-                return self.find(key)
-            except ClientError as e:
-                if e.http_status == 404:
-                    return None
-                raise e
-
+    def find_by(self, **conditions) -> Optional[Job]:
         return super().find_by(**conditions)
 
-    def reload(self) -> "Jobs":
-        """Unload the cached jobs.
 
-        Forces the next access, if any, to query the jobs from the Connect server.
-        """
-        self._cache = None
-        return self
-
-
-class JobsMixin(Resource):
+class JobsMixin(Active, Resource):
     """Mixin class to add a jobs attribute to a resource."""
 
     class HasGuid(TypedDict):
         """Has a guid."""
 
         guid: Required[str]
 
-    def __init__(self, params: ResourceParameters, **kwargs: Unpack[HasGuid]):
-        super().__init__(params, **kwargs)
-        uid = kwargs["guid"]
-        endpoint = self.params.url + f"v1/content/{uid}"
-        self.jobs = Jobs(self.params, endpoint)
+    def __init__(self, ctx, **kwargs):
+        super().__init__(ctx, **kwargs)
+        self.jobs = Jobs(Job, ctx, self)

src/posit/connect/resources.py

@@ -1,10 +1,13 @@
+import posixpath
 import warnings
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import Any, Generic, List, Optional, Type, TypeVar
+from typing import Any, Generic, List, Optional, Sequence, Type, TypeVar
 
 import requests
 
+from posit.connect.context import Context
+
 from .urls import Url
 
 
@@ -47,29 +50,74 @@ def __init__(self, params: ResourceParameters) -> None:
         self.params = params
 
 
-T = TypeVar("T", bound=Resource)
+T = TypeVar("T", bound="Active", covariant=True)
+
+
+class Active(Resource):
+    def __init__(self, ctx: Context, **kwargs):
+        params = ResourceParameters(ctx.session, ctx.url)
+        super().__init__(params, **kwargs)
+        self._ctx = ctx
 
 
-class FinderMethods(
-    Generic[T],
-    ABC,
-    Resources,
-):
-    def __init__(self, cls: Type[T], params, endpoint):
-        super().__init__(params)
+class ActiveReader(ABC, Generic[T], Sequence[T]):
+    def __init__(self, cls: Type[T], ctx: Context):
+        super().__init__()
         self._cls = cls
-        self._endpoint = endpoint
+        self._ctx = ctx
+        self._cache = None
 
     @property
     @abstractmethod
-    def _data(self) -> List[T]:
+    def _endpoint(self) -> str:
         raise NotImplementedError()
 
-    def find(self, uid):
-        endpoint = self._endpoint + str(uid)
-        response = self.params.session.get(endpoint)
-        result = response.json()
-        return self._cls(self.params, endpoint=self._endpoint, **result)
+    @property
+    def _data(self) -> List[T]:
+        if self._cache:
+            return self._cache
+
+        response = self._ctx.session.get(self._endpoint)
+        results = response.json()
+        self._cache = [self._cls(self._ctx, **result) for result in results]
+        return self._cache
+
+    def __getitem__(self, index):
+        """Retrieve an item or slice from the sequence."""
+        return self._data[index]
+
+    def __len__(self):
+        """Return the length of the sequence."""
+        return len(self._data)
+
+    def __str__(self):
+        return str(self._data)
+
+    def __repr__(self):
+        return repr(self._data)
+
+    def reload(self):
+        self._cache = None
+        return self
+
+
+class ActiveFinderMethods(ActiveReader[T], ABC, Generic[T]):
+    _uid: str = "guid"
+
+    def find(self, uid) -> T:
+        if self._cache:
+            conditions = {self._uid: uid}
+            result = self.find_by(**conditions)
+        else:
+            endpoint = posixpath.join(self._endpoint + uid)
+            response = self._ctx.session.get(endpoint)
+            result = response.json()
+            result = self._cls(self._ctx, **result)
+
+        if not result:
+            raise ValueError("")
+
+        return result
 
     def find_by(self, **conditions: Any) -> Optional[T]:
         """Finds the first record matching the specified conditions.