refactor: use static duck typing for packages (#356)

Co-authored-by: Barret Schloerke <[email protected]>

Author: tdstein

src/posit/connect/client.py

@@ -14,15 +14,15 @@
 from .groups import Groups
 from .metrics import Metrics
 from .oauth import OAuth
-from .packages import Packages
-from .resources import ResourceParameters, _ResourceSequence
+from .resources import ResourceParameters, _PaginatedResourceSequence, _ResourceSequence
 from .tags import Tags
 from .tasks import Tasks
 from .users import User, Users
 from .vanities import Vanities
 
 if TYPE_CHECKING:
     from .environments import Environments
+    from .packages import _Packages
 
 
 class Client(ContextManager):
@@ -297,9 +297,9 @@ def oauth(self) -> OAuth:
         return OAuth(self.resource_params, self.cfg.api_key)
 
     @property
-    @requires(version="2024.10.0-dev")
-    def packages(self) -> Packages:
-        return Packages(self._ctx, "v1/packages")
+    @requires(version="2024.11.0")
+    def packages(self) -> _Packages:
+        return _PaginatedResourceSequence(self._ctx, "v1/packages", uid="name")
 
     @property
     def vanities(self) -> Vanities:

src/posit/connect/content.py

@@ -20,19 +20,20 @@
 from . import tasks
 from ._api import ApiDictEndpoint, JsonifiableDict
 from .bundles import Bundles
+from .context import requires
 from .env import EnvVars
 from .errors import ClientError
 from .oauth.associations import ContentItemAssociations
-from .packages import ContentPackagesMixin as PackagesMixin
 from .permissions import Permissions
-from .resources import Resource, ResourceParameters, Resources, _ResourceSequence
+from .resources import Active, Resource, ResourceParameters, Resources, _ResourceSequence
 from .tags import ContentItemTags
 from .vanities import VanityMixin
 from .variants import Variants
 
 if TYPE_CHECKING:
     from .context import Context
     from .jobs import Jobs
+    from .packages import _ContentPackages
     from .tasks import Task
 
 
@@ -174,7 +175,7 @@ class ContentItemOwner(Resource):
     pass
 
 
-class ContentItem(PackagesMixin, VanityMixin, Resource):
+class ContentItem(Active, VanityMixin, Resource):
     class _AttrsBase(TypedDict, total=False):
         # # `name` will be set by other _Attrs classes
         # name: str
@@ -516,6 +517,12 @@ def jobs(self) -> Jobs:
         path = posixpath.join(self._path, "jobs")
         return _ResourceSequence(self._ctx, path, uid="key")
 
+    @property
+    @requires(version="2024.11.0")
+    def packages(self) -> _ContentPackages:
+        path = posixpath.join(self._path, "packages")
+        return _ResourceSequence(self._ctx, path, uid="name")
+
 
 class Content(Resources):
     """Content resource.

src/posit/connect/packages.py

@@ -1,213 +1,99 @@
 from __future__ import annotations
 
-import posixpath
-from typing import Generator, Literal, Optional, TypedDict
+from collections.abc import Mapping, Sized
+from typing import Any, Literal, Protocol, SupportsIndex, overload
 
-from typing_extensions import NotRequired, Required, Unpack
 
-from posit.connect.context import requires
-from posit.connect.errors import ClientError
-from posit.connect.paginator import Paginator
+class _ContentPackage(Mapping[str, Any]):
+    pass
 
-from .resources import Active, ActiveFinderMethods, ActiveSequence
 
+class _ContentPackages(Sized, Protocol):
+    @overload
+    def __getitem__(self, index: SupportsIndex) -> _ContentPackage: ...
 
-class ContentPackage(Active):
-    class _Package(TypedDict):
-        language: Required[str]
-        name: Required[str]
-        version: Required[str]
-        hash: Required[Optional[str]]
+    @overload
+    def __getitem__(self, index: slice) -> _ContentPackage: ...
 
-    def __init__(self, ctx, /, **attributes: Unpack[_Package]):
-        # todo - passing "" is a hack since path isn't needed. Instead, this class should inherit from Resource, but ActiveSequence is designed to operate on Active. That should change.
-        super().__init__(ctx, "", **attributes)
-
-
-class ContentPackages(ActiveFinderMethods["ContentPackage"], ActiveSequence["ContentPackage"]):
-    """A collection of packages."""
-
-    def __init__(self, ctx, path):
-        super().__init__(ctx, path, "name")
-
-    def _create_instance(self, path, /, **attributes):  # noqa: ARG002
-        return ContentPackage(self._ctx, **attributes)
-
-    def fetch(self, **conditions):
-        try:
-            return super().fetch(**conditions)
-        except ClientError as e:
-            if e.http_status == 204:
-                return []
-            raise e
-
-    def find(self, uid):
-        raise NotImplementedError("The 'find' method is not support by the Packages API.")
-
-    class _FindBy(TypedDict, total=False):
-        language: NotRequired[Literal["python", "r"]]
-        """Programming language ecosystem, options are 'python' and 'r'"""
-
-        name: NotRequired[str]
-        """The package name"""
-
-        version: NotRequired[str]
-        """The package version"""
-
-        hash: NotRequired[Optional[str]]
-        """Package description hash for R packages."""
-
-    def find_by(self, **conditions: Unpack[_FindBy]):  # type: ignore
+    def find_by(
+        self,
+        *,
+        language: Literal["python", "r"] = ...,
+        name: str = ...,
+        version: str = ...,
+        hash: str | None = ...,  # noqa: A002
+    ) -> _ContentPackage | None:
         """
         Find the first record matching the specified conditions.
 
         There is no implied ordering, so if order matters, you should specify it yourself.
 
         Parameters
         ----------
-        **conditions : Unpack[_FindBy]
-            Conditions for filtering packages. The following keys are accepted:
-
         language : {"python", "r"}, not required
             Programming language ecosystem, options are 'python' and 'r'
-
         name : str, not required
             The package name
-
         version : str, not required
             The package version
-
         hash : str or None, optional, not required
             Package description hash for R packages.
 
         Returns
         -------
-        Optional[T]
+        _ContentPackage | None
             The first record matching the specified conditions, or `None` if no such record exists.
         """
-        return super().find_by(**conditions)
-
-
-class ContentPackagesMixin(Active):
-    """Mixin class to add a packages attribute."""
-
-    @property
-    @requires(version="2024.10.0-dev")
-    def packages(self):
-        path = posixpath.join(self._path, "packages")
-        return ContentPackages(self._ctx, path)
-
-
-class Package(Active):
-    class _Package(TypedDict):
-        language: Required[Literal["python", "r"]]
-        """Programming language ecosystem, options are 'python' and 'r'"""
-
-        language_version: Required[str]
-        """Programming language version"""
-
-        name: Required[str]
-        """The package name"""
+        ...
 
-        version: Required[str]
-        """The package version"""
 
-        hash: Required[Optional[str]]
-        """Package description hash for R packages."""
+class _Package(Mapping[str, Any]):
+    pass
 
-        bundle_id: Required[str]
-        """The unique identifier of the bundle this package is associated with"""
 
-        app_id: Required[str]
-        """The numerical identifier of the application this package is associated with"""
+class _Packages(Sized, Protocol):
+    @overload
+    def __getitem__(self, index: SupportsIndex) -> _ContentPackage: ...
 
-        app_guid: Required[str]
-        """The unique identifier of the application this package is associated with"""
+    @overload
+    def __getitem__(self, index: slice) -> _ContentPackage: ...
 
-    def __init__(self, ctx, /, **attributes: Unpack[_Package]):
-        # todo - passing "" is a hack since path isn't needed. Instead, this class should inherit from Resource, but ActiveSequence is designed to operate on Active. That should change.
-        super().__init__(ctx, "", **attributes)
-
-
-class Packages(ActiveFinderMethods["Package"], ActiveSequence["Package"]):
-    def __init__(self, ctx, path):
-        super().__init__(ctx, path, "name")
-
-    def _create_instance(self, path, /, **attributes):  # noqa: ARG002
-        return Package(self._ctx, **attributes)
-
-    class _Fetch(TypedDict, total=False):
-        language: Required[Literal["python", "r"]]
-        """Programming language ecosystem, options are 'python' and 'r'"""
-
-        name: Required[str]
-        """The package name"""
-
-        version: Required[str]
-        """The package version"""
-
-    def fetch(self, **conditions: Unpack[_Fetch]) -> Generator["Package"]:  # type: ignore
-        # todo - add pagination support to ActiveSequence
-        url = self._ctx.url + self._path
-        paginator = Paginator(self._ctx.session, url, dict(**conditions))
-        for page in paginator.fetch_pages():
-            results = page.results
-            yield from (self._create_instance("", **result) for result in results)
-
-    def find(self, uid):
-        raise NotImplementedError("The 'find' method is not support by the Packages API.")
-
-    class _FindBy(TypedDict, total=False):
-        language: NotRequired[Literal["python", "r"]]
-        """Programming language ecosystem, options are 'python' and 'r'"""
-
-        language_version: NotRequired[str]
-        """Programming language version"""
-
-        name: NotRequired[str]
-        """The package name"""
-
-        version: NotRequired[str]
-        """The package version"""
-
-        hash: NotRequired[Optional[str]]
-        """Package description hash for R packages."""
-
-        bundle_id: NotRequired[str]
-        """The unique identifier of the bundle this package is associated with"""
-
-        app_id: NotRequired[str]
-        """The numerical identifier of the application this package is associated with"""
-
-        app_guid: NotRequired[str]
-        """The unique identifier of the application this package is associated with"""
-
-    def find_by(self, **conditions: Unpack[_FindBy]) -> "Package | None":  # type: ignore
+    def find_by(
+        self,
+        *,
+        language: Literal["python", "r"] = ...,
+        name: str = ...,
+        version: str = ...,
+        hash: str | None = ...,  # noqa: A002,
+        bundle_id: str = ...,
+        app_id: str = ...,
+        app_guid: str = ...,
+    ) -> _ContentPackage | None:
         """
         Find the first record matching the specified conditions.
 
         There is no implied ordering, so if order matters, you should specify it yourself.
 
         Parameters
         ----------
-        **conditions : Unpack[_FindBy]
-            Conditions for filtering packages. The following keys are accepted:
-
         language : {"python", "r"}, not required
             Programming language ecosystem, options are 'python' and 'r'
-
         name : str, not required
             The package name
-
         version : str, not required
             The package version
-
         hash : str or None, optional, not required
             Package description hash for R packages.
+        bundle_id: str, not required
+            The unique identifier of the bundle this package is associated with.
+        app_id: str, not required
+            The numerical identifier of the application this package is associated with.
+        app_guid: str, not required
+            The unique identifier of the application this package is associated with.
 
         Returns
         -------
-        Optional[Package]
+        _Package | None
             The first record matching the specified conditions, or `None` if no such record exists.
         """
-        return super().find_by(**conditions)
+        ...