feat: add packages attribute to content

src/posit/connect/content.py

@@ -11,6 +11,7 @@
 from .bundles import Bundles
 from .env import EnvVars
 from .oauth.associations import ContentItemAssociations
+from .packages import PackagesMixin
 from .permissions import Permissions
 from .resources import Resource, ResourceParameters, Resources
 from .tasks import Task
@@ -32,7 +33,7 @@ class ContentItemOwner(Resource):
  pass
 
 
-class ContentItem(VanityMixin, Resource):
+class ContentItem(PackagesMixin, VanityMixin, 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'})"
+ )