--wip-- [skip ci]

src/posit/connect/bundles.py

@@ -5,7 +5,7 @@
 
 from typing import List
 
-from . import config, resources, urls
+from . import config, resources, tasks, urls
 
 
 class BundleMetadata(resources.Resource):
@@ -96,6 +96,21 @@ def delete(self) -> None:
         url = urls.append(self.config.url, path)
         self.session.delete(url)
 
+    def deploy(self) -> tasks.Task:
+        """Deploy the bundle.
+
+        Returns
+        -------
+        tasks.Task
+            The task for the deployment.
+        """
+        path = f"v1/content/{self.content_guid}/deploy"
+        url = urls.append(self.config.url, path)
+        response = self.session.post(url, json={"bundle_id": self.id})
+        result = response.json()
+        ts = tasks.Tasks(self.config, self.session)
+        return ts.get(result["task_id"])
+
     def download(self, output: io.BufferedWriter | str):
         """Download a bundle.
 

src/posit/connect/content.py

@@ -7,7 +7,7 @@
 
 from requests import Session
 
-from . import urls
+from . import tasks, urls
 
 from .config import Config
 from .bundles import Bundles
@@ -218,6 +218,21 @@ def delete(self) -> None:
         url = urls.append(self.config.url, path)
         self.session.delete(url)
 
+    def deploy(self) -> tasks.Task:
+        """Deploy the content.
+
+        Returns
+        -------
+        tasks.Task
+            The task for the deployment.
+        """
+        path = f"v1/content/{self.guid}/deploy"
+        url = urls.append(self.config.url, path)
+        response = self.session.post(url, json={"bundle_id": None})
+        result = response.json()
+        ts = tasks.Tasks(self.config, self.session)
+        return ts.get(result["task_id"])
+
     @overload
     def update(
         self,

src/posit/connect/tasks.py

@@ -0,0 +1,174 @@
+from __future__ import annotations
+
+from typing import List, overload
+
+from . import resources, urls
+
+
+class Task(resources.Resource):
+    @property
+    def id(self) -> str:
+        """The task identifier.
+
+        Returns
+        -------
+        str
+        """
+        return self["id"]
+
+    @property
+    def is_finished(self) -> bool:
+        """The task state.
+
+        If True, the task has completed. The task may have exited successfully
+        or have failed. Inspect the exit_code to determine if the task finished
+        successfully or not.
+
+        Returns
+        -------
+        bool
+        """
+        return self.get("finished", False)
+
+    @property
+    def output(self) -> List[str]:
+        """Process output.
+
+        The process output produced by the task.
+
+        Returns
+        -------
+        List[str]
+        """
+        return self["output"]
+
+    @property
+    def error_code(self) -> int | None:
+        """The error code.
+
+        The error code produced by the task. A non-zero value represent an
+        error. A zero value represents no error.
+
+        Returns
+        -------
+        int | None
+            Non-zero value indicates an error.
+        """
+        return self["code"] if self.is_finished else None
+
+    @property
+    def error_message(self) -> str | None:
+        """The error message.
+
+        Returns
+        -------
+        str | None
+            Human readable error message, or None on success or not finished.
+        """
+        return self.get("error") if self.is_finished else None
+
+    @property
+    def result(self) -> dict | None:
+        """The task result.
+
+        Returns
+        -------
+        dict | None
+        """
+        return self.get("result")
+
+    # CRUD Methods
+
+    @overload
+    def update(self, first: int, wait: int, **kwargs) -> None:
+        """Update the task.
+
+        Parameters
+        ----------
+        first : int, default 0
+            Line to start output on.
+        wait : int, default 0
+            Maximum number of seconds to wait for the task to complete.
+        """
+        ...
+
+    @overload
+    def update(self, *args, **kwargs) -> None:
+        """Update the task."""
+        ...
+
+    def update(self, *args, **kwargs) -> None:
+        """Update the task."""
+        params = dict(*args, **kwargs)
+        path = f"v1/tasks/{self.id}"
+        url = urls.append(self.config.url, path)
+        response = self.session.get(url, params=params)
+        result = response.json()
+        super().update(**result)
+
+    def wait_for(self, sleep: int = 1) -> None:
+        """Wait for the task to finish.
+
+        Parameters
+        ----------
+        sleep : int, optional
+            Maximum number of seconds to wait between task status checks.
+        """
+        while not self.is_finished:
+            self.update(wait=sleep)
+
+
+class Tasks(resources.Resources):
+    @overload
+    def get(self, id: str, first: int, wait: int) -> Task:
+        """Get a task.
+
+        Parameters
+        ----------
+        id : str
+            Task identifier.
+        first : int, default 0
+            Line to start output on.
+        wait : int, default 0
+            Maximum number of seconds to wait for the task to complete.
+
+        Returns
+        -------
+        Task
+            _description_
+        """
+        ...
+
+    @overload
+    def get(self, id: str, *args, **kwargs) -> Task:
+        """Get a task.
+
+        Parameters
+        ----------
+        id : str
+            Task identifier.
+
+        Returns
+        -------
+        Task
+        """
+        ...
+
+    def get(self, id: str, *args, **kwargs) -> Task:
+        """Get a task.
+
+        Parameters
+        ----------
+        id : str
+            Task identifier.
+
+        Returns
+        -------
+        Task
+        """
+        params = dict(*args, **kwargs)
+        path = f"v1/tasks/{id}"
+        url = urls.append(self.config.url, path)
+        response = self.session.get(url, params=params)
+        result = response.json()
+        return Task(self.config, self.session, **result)