feat: adds bundle create and download (#179)

Author: tdstein

src/posit/connect/bundles.py

@@ -1,17 +1,14 @@
 from __future__ import annotations
+import io
 
-from typing import List
-
-from requests.sessions import Session as Session
+import requests
 
-from posit.connect.config import Config
-
-from . import urls
+from typing import List
 
-from .resources import Resources, Resource
+from . import config, resources, urls
 
 
-class BundleMetadata(Resource):
+class BundleMetadata(resources.Resource):
     @property
     def source(self) -> str | None:
         return self.get("source")
@@ -37,7 +34,7 @@ def archive_sha1(self) -> str | None:
         return self.get("archive_sha1")
 
 
-class Bundle(Resource):
+class Bundle(resources.Resource):
     @property
     def id(self) -> str:
         return self["id"]
@@ -99,14 +96,115 @@ def delete(self) -> None:
         url = urls.append(self.config.url, path)
         self.session.delete(url)
 
+    def download(self, output: io.BufferedWriter | str):
+        """Download a bundle.
+
+        Download a bundle to a file or memory.
+
+        Parameters
+        ----------
+        output: io.BufferedWriter | str
+            An io.BufferedWriter instance or a str representing a relative or absolute path.
+
+        Raises
+        ------
+        TypeError
+            If the output is not of type `io.BufferedWriter` or `str`.
+
+        Examples
+        --------
+        Write to a file.
+        >>> bundle.download("bundle.tar.gz")
+        None
+
+        Write to an io.BufferedWriter.
+        >>> with open('bundle.tar.gz', 'wb') as file:
+        >>>     bundle.download(file)
+        None
+        """
+        if not isinstance(output, (io.BufferedWriter, str)):
+            raise TypeError(
+                f"download() expected argument type 'io.BufferedWriter` or 'str', but got '{type(input).__name__}'"
+            )
+
+        path = f"v1/content/{self.content_guid}/bundles/{self.id}/download"
+        url = urls.append(self.config.url, path)
+        response = self.session.get(url, stream=True)
+        if isinstance(output, io.BufferedWriter):
+            for chunk in response.iter_content():
+                output.write(chunk)
+            return
+
+        if isinstance(output, str):
+            with open(output, "wb") as file:
+                for chunk in response.iter_content():
+                    file.write(chunk)
+            return
 
-class Bundles(Resources):
+
+class Bundles(resources.Resources):
     def __init__(
-        self, config: Config, session: Session, content_guid: str
+        self,
+        config: config.Config,
+        session: requests.Session,
+        content_guid: str,
     ) -> None:
         super().__init__(config, session)
         self.content_guid = content_guid
 
+    def create(self, input: io.BufferedReader | bytes | str) -> Bundle:
+        """Create a bundle.
+
+        Create a bundle from a file or memory.
+
+        Parameters
+        ----------
+        input : io.BufferedReader | bytes | str
+            Input archive for bundle creation. A 'str' type assumes a relative or absolute filepath.
+
+        Returns
+        -------
+        Bundle
+            The created bundle.
+
+        Raises
+        ------
+        TypeError
+            If the input is not of type `io.BufferedReader`, `bytes`, or `str`.
+
+        Examples
+        --------
+        Create a bundle from io.BufferedReader
+        >>> with open('bundle.tar.gz', 'rb') as file:
+        >>>     bundle.create(file)
+        None
+
+        Create a bundle from bytes.
+        >>> with open('bundle.tar.gz', 'rb') as file:
+        >>>     data: bytes = file.read()
+        >>>     bundle.create(data)
+        None
+
+        Create a bundle from pathname.
+        >>> bundle.create("bundle.tar.gz")
+        None
+        """
+        if isinstance(input, (io.BufferedReader, bytes)):
+            data = input
+        elif isinstance(input, str):
+            with open(input, "rb") as file:
+                data = file.read()
+        else:
+            raise TypeError(
+                f"create() expected argument type 'io.BufferedReader', 'bytes', or 'str', but got '{type(input).__name__}'"
+            )
+
+        path = f"v1/content/{self.content_guid}/bundles"
+        url = urls.append(self.config.url, path)
+        response = self.session.post(url, data=data)
+        result = response.json()
+        return Bundle(self.config, self.session, **result)
+
     def find(self) -> List[Bundle]:
         path = f"v1/content/{self.content_guid}/bundles"
         url = urls.append(self.config.url, path)

tests/posit/connect/__api__/v1/content/f2f37341-e21d-3d80-c698-a935ad614066/bundles/101/download/bundle.tar.gz

@@ -29,3 +29,7 @@ def load_mock(path: str) -> dict:
     >>> data = load_mock("v1/example.jsonc")
     """
     return json.loads((Path(__file__).parent / "__api__" / path).read_text())
+
+
+def get_path(path: str) -> Path:
+    return Path(__file__).parent / "__api__" / path

tests/posit/connect/api.py

@@ -1,11 +1,16 @@
+import io
+
+import pytest
 import requests
 import responses
 
+from unittest import mock
+
 from posit.connect import Client
 from posit.connect.config import Config
 from posit.connect.bundles import Bundle
 
-from .api import load_mock  # type: ignore
+from .api import load_mock, get_path  # type: ignore
 
 
 class TestBundleProperties:
@@ -119,6 +124,216 @@ def test(self):
         assert mock_bundle_delete.call_count == 1
 
 
+class TestBundleDownload:
+    @mock.patch("builtins.open", new_callable=mock.mock_open)
+    @responses.activate
+    def test_output_as_str(self, mock_file: mock.MagicMock):
+        content_guid = "f2f37341-e21d-3d80-c698-a935ad614066"
+        bundle_id = "101"
+        path = get_path(
+            f"v1/content/{content_guid}/bundles/{bundle_id}/download/bundle.tar.gz"
+        )
+
+        # behavior
+        mock_content_get = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}",
+            json=load_mock(f"v1/content/{content_guid}.json"),
+        )
+
+        mock_bundle_get = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}/bundles/{bundle_id}",
+            json=load_mock(
+                f"v1/content/{content_guid}/bundles/{bundle_id}.json"
+            ),
+        )
+
+        mock_bundle_download = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}/bundles/{bundle_id}/download",
+            body=path.read_bytes(),
+        )
+
+        # setup
+        c = Client("12345", "https://connect.example")
+        bundle = c.content.get(content_guid).bundles.get(bundle_id)
+
+        # invoke
+        bundle.download("pathname")
+
+        # assert
+        assert mock_content_get.call_count == 1
+        assert mock_bundle_get.call_count == 1
+        assert mock_bundle_download.call_count == 1
+        mock_file.assert_called_once_with("pathname", "wb")
+
+    @responses.activate
+    def test_output_as_io(self):
+        content_guid = "f2f37341-e21d-3d80-c698-a935ad614066"
+        bundle_id = "101"
+        path = get_path(
+            f"v1/content/{content_guid}/bundles/{bundle_id}/download/bundle.tar.gz"
+        )
+
+        # behavior
+        mock_content_get = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}",
+            json=load_mock(f"v1/content/{content_guid}.json"),
+        )
+
+        mock_bundle_get = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}/bundles/{bundle_id}",
+            json=load_mock(
+                f"v1/content/{content_guid}/bundles/{bundle_id}.json"
+            ),
+        )
+
+        mock_bundle_download = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}/bundles/{bundle_id}/download",
+            body=path.read_bytes(),
+        )
+
+        # setup
+        c = Client("12345", "https://connect.example")
+        bundle = c.content.get(content_guid).bundles.get(bundle_id)
+
+        # invoke
+        file = io.BytesIO()
+        buffer = io.BufferedWriter(file)
+        bundle.download(buffer)
+        buffer.seek(0)
+
+        # assert
+        assert mock_content_get.call_count == 1
+        assert mock_bundle_get.call_count == 1
+        assert mock_bundle_download.call_count == 1
+        assert file.read() == path.read_bytes()
+
+    @responses.activate
+    def test_invalid_arguments(self):
+        content_guid = "f2f37341-e21d-3d80-c698-a935ad614066"
+        bundle_id = "101"
+        path = get_path(
+            f"v1/content/{content_guid}/bundles/{bundle_id}/download/bundle.tar.gz"
+        )
+
+        # behavior
+        mock_content_get = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}",
+            json=load_mock(f"v1/content/{content_guid}.json"),
+        )
+
+        mock_bundle_get = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}/bundles/{bundle_id}",
+            json=load_mock(
+                f"v1/content/{content_guid}/bundles/{bundle_id}.json"
+            ),
+        )
+
+        mock_bundle_download = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}/bundles/{bundle_id}/download",
+            body=path.read_bytes(),
+        )
+
+        # setup
+        c = Client("12345", "https://connect.example")
+        bundle = c.content.get(content_guid).bundles.get(bundle_id)
+
+        # invoke
+        with pytest.raises(TypeError):
+            bundle.download(None)
+
+        # assert
+        assert mock_content_get.call_count == 1
+        assert mock_bundle_get.call_count == 1
+
+
+class TestBundlesCreate:
+    @responses.activate
+    def test(self):
+        content_guid = "f2f37341-e21d-3d80-c698-a935ad614066"
+        bundle_id = "101"
+        pathname = get_path(
+            f"v1/content/{content_guid}/bundles/{bundle_id}/download/bundle.tar.gz"
+        )
+
+        # behavior
+        mock_content_get = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}",
+            json=load_mock(f"v1/content/{content_guid}.json"),
+        )
+
+        mock_bundle_post = responses.post(
+            f"https://connect.example/__api__/v1/content/{content_guid}/bundles",
+            json=load_mock(
+                f"v1/content/{content_guid}/bundles/{bundle_id}.json"
+            ),
+        )
+
+        # setup
+        c = Client("12345", "https://connect.example")
+        content = c.content.get(content_guid)
+
+        # invoke
+        data = pathname.read_bytes()
+        bundle = content.bundles.create(data)
+
+        # # assert
+        assert bundle.id == "101"
+        assert mock_content_get.call_count == 1
+        assert mock_bundle_post.call_count == 1
+
+    @responses.activate
+    def test_kwargs_pathname(self):
+        content_guid = "f2f37341-e21d-3d80-c698-a935ad614066"
+        bundle_id = "101"
+        pathname = get_path(
+            f"v1/content/{content_guid}/bundles/{bundle_id}/download/bundle.tar.gz"
+        )
+
+        # behavior
+        mock_content_get = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}",
+            json=load_mock(f"v1/content/{content_guid}.json"),
+        )
+
+        mock_bundle_post = responses.post(
+            f"https://connect.example/__api__/v1/content/{content_guid}/bundles",
+            json=load_mock(
+                f"v1/content/{content_guid}/bundles/{bundle_id}.json"
+            ),
+        )
+
+        # setup
+        c = Client("12345", "https://connect.example")
+        content = c.content.get(content_guid)
+
+        # invoke
+        pathname = str(pathname.absolute())
+        bundle = content.bundles.create(pathname)
+
+        # # assert
+        assert bundle.id == "101"
+        assert mock_content_get.call_count == 1
+        assert mock_bundle_post.call_count == 1
+
+    @responses.activate
+    def test_invalid_arguments(self):
+        content_guid = "f2f37341-e21d-3d80-c698-a935ad614066"
+
+        # behavior
+        responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}",
+            json=load_mock(f"v1/content/{content_guid}.json"),
+        )
+
+        # setup
+        c = Client("12345", "https://connect.example")
+        content = c.content.get(content_guid)
+
+        # invoke
+        with pytest.raises(TypeError):
+            content.bundles.create(None)
+
+
 class TestBundlesFind:
     @responses.activate
     def test(self):