iframe-proxy | Sunbelt Computer Software

555 lines (438 loc) · 15 KB
from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, TypedDict, Union
from typing_extensions import Unpack, deprecated
from replicate.account import Account
from replicate.helpers import async_encode_json, encode_json
from replicate.pagination import Page
from replicate.prediction import (
    Prediction,
    _create_prediction_body,
    _create_prediction_request_params,
    _json_to_prediction,
from replicate.resource import Namespace, Resource
    from pydantic import v1 as pydantic  # type: ignore
except ImportError:
    import pydantic  # type: ignore
if TYPE_CHECKING:
    from replicate.client import Client
    from replicate.prediction import Predictions
class Deployment(Resource):
    A deployment of a model hosted on Replicate.
    _client: "Client" = pydantic.PrivateAttr()
    owner: str
    The name of the user or organization that owns the deployment.
    name: str
    The name of the deployment.
    class Release(Resource):
        """
        A release of a deployment.
        """
        number: int
        """
        The release number.
        """
        model: str
        """
        The model identifier string in the format of `{model_owner}/{model_name}`.
        """
        version: str
        """
        The ID of the model version used in the release.
        """
        created_at: str
        """
        The time the release was created.
        """
        created_by: Optional[Account]
        """
        The account that created the release.
        """
        class Configuration(Resource):
            A configuration for a deployment.
            hardware: str
            The SKU for the hardware used to run the model.
            min_instances: int
            The minimum number of instances for scaling.
            max_instances: int
            The maximum number of instances for scaling.
        configuration: Configuration
        """
        The deployment configuration.
        """
    current_release: Optional[Release]
    The current release of the deployment.
    @property
    @deprecated("Use `deployment.owner` instead.")
    def username(self) -> str:
        """
        The name of the user or organization that owns the deployment.
        This attribute is deprecated and will be removed in future versions.
        """
        return self.owner
    @property
    def id(self) -> str:
        """
        Return the qualified deployment name, in the format `owner/name`.
        """
        return f"{self.owner}/{self.name}"
    @property
    def predictions(self) -> "DeploymentPredictions":
        """
        Get the predictions for this deployment.
        """
        return DeploymentPredictions(client=self._client, deployment=self)
class Deployments(Namespace):
    Namespace for operations related to deployments.
    _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.
        Args:
            name: The name of the deployment, in the format `owner/model-name`.
        Returns:
            The model.
        """
        owner, name = name.split("/", 1)
        resp = self._client._request(
            "GET",
            f"/v1/deployments/{owner}/{name}",
        return _json_to_deployment(self._client, resp.json())
    async def async_get(self, name: str) -> Deployment:
        """
        Get a deployment by name.
        Args:
            name: The name of the deployment, in the format `owner/model-name`.
        Returns:
            The model.
        """
        owner, name = name.split("/", 1)
        resp = await self._client._async_request(
            "GET",
            f"/v1/deployments/{owner}/{name}",
        return _json_to_deployment(self._client, resp.json())
    class CreateDeploymentParams(TypedDict):
        """
        Parameters for creating a new deployment.
        """
        name: str
        """The name of the deployment."""
        model: str
        """The model identifier string in the format of `{model_owner}/{model_name}`."""
        version: str
        """The version of the model to deploy."""
        hardware: str
        """The SKU for the hardware used to run the model."""
        min_instances: int
        """The minimum number of instances for scaling."""
        max_instances: int
        """The maximum number of instances for scaling."""
    def create(self, **params: Unpack[CreateDeploymentParams]) -> Deployment:
        """
        Create a new deployment.
        Args:
            params: Configuration for the new deployment.
        Returns:
            The newly created Deployment.
        """
        if name := params.get("name", None):
            if "/" in name:
                _, name = name.split("/", 1)
            params["name"] = name
        resp = self._client._request(
            "POST",
            "/v1/deployments",
            json=params,
        return _json_to_deployment(self._client, resp.json())
    async def async_create(
        self, **params: Unpack[CreateDeploymentParams]
    ) -> Deployment:
        """
        Create a new deployment.
        Args:
            params: Configuration for the new deployment.
        Returns:
            The newly created Deployment.
        """
        if name := params.get("name", None):
            if "/" in name:
                _, name = name.split("/", 1)
            params["name"] = name
        resp = await self._client._async_request(
            "POST",
            "/v1/deployments",
            json=params,
        return _json_to_deployment(self._client, resp.json())
    class UpdateDeploymentParams(TypedDict, total=False):
        """
        Parameters for updating an existing deployment.
        """
        version: str
        """The version of the model to deploy."""
        hardware: str
        """The SKU for the hardware used to run the model."""
        min_instances: int
        """The minimum number of instances for scaling."""
        max_instances: int
        """The maximum number of instances for scaling."""
    def update(
        self,
        deployment_owner: str,
        deployment_name: str,
        **params: Unpack[UpdateDeploymentParams],
    ) -> Deployment:
        """
        Update an existing deployment.
        Args:
            deployment_owner: The owner of the deployment.
            deployment_name: The name of the deployment.
            params: Configuration updates for the deployment.
        Returns:
            The updated Deployment.
        """
        resp = self._client._request(
            "PATCH",
            f"/v1/deployments/{deployment_owner}/{deployment_name}",
            json=params,
        return _json_to_deployment(self._client, resp.json())
    async def async_update(
        self,
        deployment_owner: str,
        deployment_name: str,
        **params: Unpack[UpdateDeploymentParams],
    ) -> Deployment:
        """
        Update an existing deployment.
        Args:
            deployment_owner: The owner of the deployment.
            deployment_name: The name of the deployment.
            params: Configuration updates for the deployment.
        Returns:
            The updated Deployment.
        """
        resp = await self._client._async_request(
            "PATCH",
            f"/v1/deployments/{deployment_owner}/{deployment_name}",
            json=params,
        return _json_to_deployment(self._client, resp.json())
    def delete(self, deployment_owner: str, deployment_name: str) -> bool:
        """
        Delete an existing deployment.
        Args:
            deployment_owner: The owner of the deployment.
            deployment_name: The name of the deployment.
        """
        resp = self._client._request(
            "DELETE",
            f"/v1/deployments/{deployment_owner}/{deployment_name}",
        return resp.status_code == 204
    async def async_delete(self, deployment_owner: str, deployment_name: str) -> bool:
        """
        Delete an existing deployment asynchronously.
        Args:
            deployment_owner: The owner of the deployment.
            deployment_name: The name of the deployment.
        """
        resp = await self._client._async_request(
            "DELETE",
            f"/v1/deployments/{deployment_owner}/{deployment_name}",
        return resp.status_code == 204
    @property
    def predictions(self) -> "DeploymentsPredictions":
        """
        Get predictions for deployments.
        """
        return DeploymentsPredictions(client=self._client)
def _json_to_deployment(client: "Client", json: Dict[str, Any]) -> Deployment:
    deployment = Deployment(**json)
    deployment._client = client
    return deployment
class DeploymentPredictions(Namespace):
    Namespace for operations related to predictions in a deployment.
    _deployment: Deployment
    def __init__(self, client: "Client", deployment: Deployment) -> None:
        super().__init__(client=client)
        self._deployment = deployment
    def create(
        self,
        input: Dict[str, Any],
        **params: Unpack["Predictions.CreatePredictionParams"],
    ) -> Prediction:
        """
        Create a new prediction with the deployment.
        """
        wait = params.pop("wait", None)
        file_encoding_strategy = params.pop("file_encoding_strategy", None)
        if input is not None:
            input = encode_json(
                input,
                client=self._client,
                file_encoding_strategy=file_encoding_strategy,
        body = _create_prediction_body(version=None, input=input, **params)
        extras = _create_prediction_request_params(
            wait=wait,
        resp = self._client._request(
            "POST",
            f"/v1/deployments/{self._deployment.owner}/{self._deployment.name}/predictions",
            json=body,
            **extras,
        return _json_to_prediction(self._client, resp.json())
    async def async_create(
        self,
        input: Dict[str, Any],
        **params: Unpack["Predictions.CreatePredictionParams"],
    ) -> Prediction:
        """
        Create a new prediction with the deployment.
        """
        wait = params.pop("wait", None)
        file_encoding_strategy = params.pop("file_encoding_strategy", None)
        if input is not None:
            input = await async_encode_json(
                input,
                client=self._client,
                file_encoding_strategy=file_encoding_strategy,
        body = _create_prediction_body(version=None, input=input, **params)
        extras = _create_prediction_request_params(
            wait=wait,
        resp = await self._client._async_request(
            "POST",
            f"/v1/deployments/{self._deployment.owner}/{self._deployment.name}/predictions",
            json=body,
            **extras,
        return _json_to_prediction(self._client, resp.json())
class DeploymentsPredictions(Namespace):
    Namespace for operations related to predictions in deployments.
    def create(
        self,
        deployment: Union[str, Tuple[str, str], Deployment],
        input: Dict[str, Any],
        **params: Unpack["Predictions.CreatePredictionParams"],
    ) -> Prediction:
        """
        Create a new prediction with the deployment.
        """
        wait = params.pop("wait", None)
        file_encoding_strategy = params.pop("file_encoding_strategy", None)
        url = _create_prediction_url_from_deployment(deployment)
        if input is not None:
            input = encode_json(
                input,
                client=self._client,
                file_encoding_strategy=file_encoding_strategy,
        body = _create_prediction_body(version=None, input=input, **params)
        extras = _create_prediction_request_params(wait=wait)
        resp = self._client._request("POST", url, json=body, **extras)
        return _json_to_prediction(self._client, resp.json())
    async def async_create(
        self,
        deployment: Union[str, Tuple[str, str], Deployment],
        input: Dict[str, Any],
        **params: Unpack["Predictions.CreatePredictionParams"],
    ) -> Prediction:
        """
        Create a new prediction with the deployment.
        """
        wait = params.pop("wait", None)
        file_encoding_strategy = params.pop("file_encoding_strategy", None)
        url = _create_prediction_url_from_deployment(deployment)
        if input is not None:
            input = await async_encode_json(
                input,
                client=self._client,
                file_encoding_strategy=file_encoding_strategy,
        body = _create_prediction_body(version=None, input=input, **params)
        extras = _create_prediction_request_params(wait=wait)
        resp = await self._client._async_request("POST", url, json=body, **extras)
        return _json_to_prediction(self._client, resp.json())
def _create_prediction_url_from_deployment(
    deployment: Union[str, Tuple[str, str], Deployment],
    owner, name = None, None
    if isinstance(deployment, Deployment):
        owner, name = deployment.owner, deployment.name
    elif isinstance(deployment, tuple):
        owner, name = deployment[0], deployment[1]
    elif isinstance(deployment, str):
        owner, name = deployment.split("/", 1)
    if owner is None or name is None:
        raise ValueError(
            "deployment must be a Deployment, a tuple of (owner, name), or a string in the format 'owner/name'"
    return f"/v1/deployments/{owner}/{name}/predictions"
Sunbelt Computer Software

PL/B Language Development and Support

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Sunbelt Computer Software

PL/B Language Development and Support

FilesExpand file tree

deployment.py

Latest commit

History

deployment.py

File metadata and controls
