Add support for deployment list, create, and update endpoints

Signed-off-by: Mattt Zmuda <[email protected]>

Author: mattt

replicate/deployment.py

@@ -1,8 +1,9 @@
-from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
+from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, TypedDict, Union
 
 from typing_extensions import Unpack, deprecated
 
 from replicate.account import Account
+from replicate.pagination import Page
 from replicate.prediction import (
     Prediction,
     _create_prediction_body,
@@ -140,6 +141,55 @@ class Deployments(Namespace):
 
     _client: "Client"
 
+    def list(
+        self,
+        cursor: Union[str, "ellipsis", None] = ...,  # noqa: F821
+    ) -> Page[Deployment]:
+        """
+        List all deployments.
+
+        Returns:
+            A page of Deployments.
+        """
+
+        if cursor is None:
+            raise ValueError("cursor cannot be None")
+
+        resp = self._client._request(
+            "GET", "/v1/deployments" if cursor is ... else cursor
+        )
+
+        obj = resp.json()
+        obj["results"] = [
+            _json_to_deployment(self._client, result) for result in obj["results"]
+        ]
+
+        return Page[Deployment](**obj)
+
+    async def async_list(
+        self,
+        cursor: Union[str, "ellipsis", None] = ...,  # noqa: F821
+    ) -> Page[Deployment]:
+        """
+        List all deployments.
+
+        Returns:
+            A page of Deployments.
+        """
+        if cursor is None:
+            raise ValueError("cursor cannot be None")
+
+        resp = await self._client._async_request(
+            "GET", "/v1/deployments" if cursor is ... else cursor
+        )
+
+        obj = resp.json()
+        obj["results"] = [
+            _json_to_deployment(self._client, result) for result in obj["results"]
+        ]
+
+        return Page[Deployment](**obj)
+
     def get(self, name: str) -> Deployment:
         """
         Get a deployment by name.
@@ -178,6 +228,126 @@ async def async_get(self, name: str) -> Deployment:
 
         return _json_to_deployment(self._client, resp.json())
 
+    class CreateDeploymentParams(TypedDict):
+        """
+        Parameters for creating a new deployment.
+        """
+
+        name: str
+        model: str
+        version: str
+        hardware: str
+        min_instances: int
+        max_instances: int
+
+    def create(self, deployment_config: CreateDeploymentParams) -> Deployment:
+        """
+        Create a new deployment.
+
+        Args:
+            deployment_config: Configuration for the new deployment.
+        Returns:
+            The newly created Deployment.
+        """
+
+        if name := deployment_config.get("name", None):
+            if "/" in name:
+                _, name = name.split("/", 1)
+            deployment_config["name"] = name
+
+        resp = self._client._request(
+            "POST",
+            "/v1/deployments",
+            json=deployment_config,
+        )
+
+        return _json_to_deployment(self._client, resp.json())
+
+    async def async_create(
+        self, deployment_config: CreateDeploymentParams
+    ) -> Deployment:
+        """
+        Create a new deployment.
+
+        Args:
+            deployment_config: Configuration for the new deployment.
+        Returns:
+            The newly created Deployment.
+        """
+
+        if name := deployment_config.get("name", None):
+            if "/" in name:
+                _, name = name.split("/", 1)
+            deployment_config["name"] = name
+
+        resp = await self._client._async_request(
+            "POST",
+            "/v1/deployments",
+            json=deployment_config,
+        )
+
+        return _json_to_deployment(self._client, resp.json())
+
+    class UpdateDeploymentParams(TypedDict, total=False):
+        """
+        Parameters for updating an existing deployment.
+        """
+
+        version: str
+        hardware: str
+        min_instances: int
+        max_instances: int
+
+    def update(
+        self,
+        deployment_owner: str,
+        deployment_name: str,
+        deployment_config: UpdateDeploymentParams,
+    ) -> Deployment:
+        """
+        Update an existing deployment.
+
+        Args:
+            deployment_owner: The owner of the deployment.
+            deployment_name: The name of the deployment.
+            deployment_config: Configuration updates for the deployment.
+        Returns:
+            The updated Deployment.
+        """
+
+        resp = self._client._request(
+            "PATCH",
+            f"/v1/deployments/{deployment_owner}/{deployment_name}",
+            json=deployment_config,
+        )
+
+        return _json_to_deployment(self._client, resp.json())
+
+    async def async_update(
+        self,
+        deployment_owner: str,
+        deployment_name: str,
+        deployment_config: UpdateDeploymentParams,
+    ) -> Deployment:
+        """
+        Update an existing deployment.
+
+        Args:
+            deployment_owner: The owner of the deployment.
+            deployment_name: The name of the deployment.
+            deployment_config: Configuration updates for the deployment.
+        Returns:
+            The updated Deployment.
+        """
+
+        resp = await self._client._async_request(
+            "PATCH",
+            f"/v1/deployments/{deployment_owner}/{deployment_name}",
+            json=deployment_config,
+        )
+
+        return _json_to_deployment(self._client, resp.json())
+
     @property
     def predictions(self) -> "DeploymentsPredictions":
         """