feat: add packages attribute to content

Author: tdstein

src/posit/connect/content.py

@@ -8,6 +8,7 @@
 from typing import Any, List, Literal, Optional, overload
 
 from posit.connect.oauth.associations import ContentItemAssociations
+from posit.connect.packages import PackagesMixin
 
 from . import tasks
 from .bundles import Bundles
@@ -32,7 +33,7 @@ class ContentItemOwner(Resource):
     pass
 
 
-class ContentItem(Resource):
+class ContentItem(PackagesMixin, Resource):
     def __getitem__(self, key: Any) -> Any:
         v = super().__getitem__(key)
         if key == "owner" and isinstance(v, dict):

src/posit/connect/packages.py

@@ -0,0 +1,89 @@
+from typing import Literal, Optional, Sequence, TypedDict
+
+from typing_extensions import NotRequired, Required, Unpack
+
+from .resources import Resource, ResourceParameters, Resources
+
+
+class Package(Resource):
+    """A package resource."""
+
+    class PackageAttributes(TypedDict):
+        """Package attributes."""
+
+        language: Required[Literal["r", "python"]]
+        name: Required[str]
+        version: Required[str]
+        hash: NotRequired[str]
+
+    def __init__(self, params: ResourceParameters, **kwargs: Unpack[PackageAttributes]):
+        super().__init__(params, **kwargs)
+
+
+class Packages(Resources, Sequence[Package]):
+    """A collection of packages."""
+
+    def __init__(self, params, endpoint):
+        super().__init__(params)
+        self._endpoint = endpoint
+        self._packages = []
+        self.reload()
+
+    def __getitem__(self, index):
+        """Retrieve an item or slice from the sequence."""
+        return self._packages[index]
+
+    def __len__(self):
+        """Return the length of the sequence."""
+        return len(self._packages)
+
+    def __repr__(self):
+        """Return the string representation of the sequence."""
+        return f"Packages({', '.join(map(str, self._packages))})"
+
+    def count(self, value):
+        """Return the number of occurrences of a value in the sequence."""
+        return self._packages.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._packages)
+        return self._packages.index(value, start, stop)
+
+    def reload(self) -> "Packages":
+        """Reload packages from the Connect server.
+
+        Returns
+        -------
+        List[Package]
+        """
+        response = self.params.session.get(self._endpoint)
+        results = response.json()
+        packages = [Package(self.params, **result) for result in results]
+        self._packages = packages
+        return self
+
+
+class PackagesMixin(Resource):
+    """Mixin class to add a packages to a resource."""
+
+    class HasGuid(TypedDict):
+        """Has a guid."""
+
+        guid: Required[str]
+
+    def __init__(self, params: ResourceParameters, **kwargs: Unpack[HasGuid]):
+        super().__init__(params, **kwargs)
+        self._guid = kwargs["guid"]
+        self._packages: Optional[Packages] = None
+
+    @property
+    def packages(self) -> Packages:
+        """Get the packages."""
+        if self._packages:
+            return self._packages
+
+        endpoint = self.params.url + f"v1/content/{self._guid}/packages"
+        self._packages = Packages(self.params, endpoint)
+        return self._packages

src/posit/connect/paginator.py

@@ -38,7 +38,7 @@ class Paginator:
         url (str): The URL of the paginated API endpoint.
     """
 
-    def __init__(self, session: requests.Session, url: str, params = {}) -> None:
+    def __init__(self, session: requests.Session, url: str, params={}) -> None:
         self.session = session
         self.url = url
         self.params = params

tests/posit/connect/test_packages.py

@@ -0,0 +1,128 @@
+import requests
+import responses
+
+from posit.connect.packages import PackagesMixin
+from posit.connect.resources import ResourceParameters
+from posit.connect.urls import Url
+
+
+class TestPackagesMixin:
+    def setup_method(self):
+        self.url = Url("http://connect.example/__api__")
+        self.endpoint = self.url + "v1/content/1/packages"
+        self.session = requests.Session()
+        self.params = ResourceParameters(self.session, self.url)
+        self.mixin = PackagesMixin(self.params, guid="1")
+
+    @responses.activate
+    def test_packages(self):
+        # mock
+        mock_get = responses.get(
+            self.endpoint,
+            json=[
+                {
+                    "language": "python",
+                    "name": "posit-sdk",
+                    "version": "0.5.1.dev3+gd4bba40.d20241016",
+                }
+            ],
+        )
+
+        # call
+        packages = self.mixin.packages
+
+        # assert
+        assert mock_get.call_count == 1
+        assert packages[0] == {
+            "language": "python",
+            "name": "posit-sdk",
+            "version": "0.5.1.dev3+gd4bba40.d20241016",
+        }
+
+    @responses.activate
+    def test_packages_are_cached(self):
+        # mock
+        mock_get = responses.get(
+            self.endpoint,
+            json=[
+                {
+                    "language": "python",
+                    "name": "posit-sdk",
+                    "version": "0.5.1.dev3+gd4bba40.d20241016",
+                }
+            ],
+        )
+
+        # call attribute twice, the second call should be cached
+        self.mixin.packages
+        self.mixin.packages
+
+        # assert called once
+        assert mock_get.call_count == 1
+
+    @responses.activate
+    def test_packages_count(self):
+        responses.get(
+            self.endpoint,
+            json=[
+                {
+                    "language": "python",
+                    "name": "posit-sdk",
+                    "version": "0.5.1.dev3+gd4bba40.d20241016",
+                }
+            ],
+        )
+
+        packages = self.mixin.packages
+        count = packages.count(
+            {
+                "language": "python",
+                "name": "posit-sdk",
+                "version": "0.5.1.dev3+gd4bba40.d20241016",
+            }
+        )
+
+        assert count == 1
+
+    @responses.activate
+    def test_packages_index(self):
+        responses.get(
+            self.endpoint,
+            json=[
+                {
+                    "language": "python",
+                    "name": "posit-sdk",
+                    "version": "0.5.1.dev3+gd4bba40.d20241016",
+                }
+            ],
+        )
+
+        packages = self.mixin.packages
+        index = packages.index(
+            {
+                "language": "python",
+                "name": "posit-sdk",
+                "version": "0.5.1.dev3+gd4bba40.d20241016",
+            }
+        )
+
+        assert index == 0
+
+    @responses.activate
+    def test_packages_repr(self):
+        responses.get(
+            self.endpoint,
+            json=[
+                {
+                    "language": "python",
+                    "name": "posit-sdk",
+                    "version": "0.5.1.dev3+gd4bba40.d20241016",
+                }
+            ],
+        )
+
+        packages = self.mixin.packages
+        assert (
+            repr(packages)
+            == "Packages({'language': 'python', 'name': 'posit-sdk', 'version': '0.5.1.dev3+gd4bba40.d20241016'})"
+        )