# This file was auto-generated by Fern from our API Definition. import typing import urllib.parse from json.decoder import JSONDecodeError from ...core.api_error import ApiError from ...core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ...core.jsonable_encoder import jsonable_encoder from ...core.remove_none_from_dict import remove_none_from_dict from ...errors.unprocessable_entity_error import UnprocessableEntityError from ...types.deployment_history_response import DeploymentHistoryResponse from ...types.deployment_response import DeploymentResponse from ...types.deployments_list_response import DeploymentsListResponse from ...types.http_validation_error import HttpValidationError from ...types.projects_list_response import ProjectsListResponse from ...types.repository_validation_response import RepositoryValidationResponse from ...types.version_response import VersionResponse try: import pydantic if pydantic.__version__.startswith("1."): raise ImportError import pydantic.v1 as pydantic # type: ignore except ImportError: import pydantic # type: ignore # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class AgentsDeploymentsControlPlaneClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def get_deployments( self, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None ) -> DeploymentsListResponse: """ Parameters: - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.agents_deployments_control_plane.get_deployments() """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1beta1/deployments"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(DeploymentsListResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def create_deployment( self, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, deployment_file_path: typing.Optional[str] = OMIT, git_ref: typing.Optional[str] = OMIT, llama_deploy_version: typing.Optional[str] = OMIT, name: str, personal_access_token: typing.Optional[str] = OMIT, repo_url: str, secrets: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT, ) -> DeploymentResponse: """ Parameters: - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. - deployment_file_path: typing.Optional[str]. - git_ref: typing.Optional[str]. - llama_deploy_version: typing.Optional[str]. - name: str. - personal_access_token: typing.Optional[str]. - repo_url: str. - secrets: typing.Optional[typing.Dict[str, typing.Optional[str]]]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.agents_deployments_control_plane.create_deployment( name="string", repo_url="string", ) """ _request: typing.Dict[str, typing.Any] = {"name": name, "repo_url": repo_url} if deployment_file_path is not OMIT: _request["deployment_file_path"] = deployment_file_path if git_ref is not OMIT: _request["git_ref"] = git_ref if llama_deploy_version is not OMIT: _request["llama_deploy_version"] = llama_deploy_version if personal_access_token is not OMIT: _request["personal_access_token"] = personal_access_token if secrets is not OMIT: _request["secrets"] = secrets _response = self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1beta1/deployments"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), json=jsonable_encoder(_request), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(DeploymentResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def get_version(self) -> VersionResponse: """ from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.agents_deployments_control_plane.get_version() """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1beta1/deployments-public/version"), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(VersionResponse, _response.json()) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def get_projects(self) -> ProjectsListResponse: """ Get all unique projects with their deployment counts --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.agents_deployments_control_plane.get_projects() """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1beta1/deployments/list-projects"), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(ProjectsListResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def validate_repository( self, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, deployment_id: typing.Optional[str] = OMIT, pat: typing.Optional[str] = OMIT, repository_url: str, ) -> RepositoryValidationResponse: """ Validate repository access and return unified response. Parameters: - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. - deployment_id: typing.Optional[str]. - pat: typing.Optional[str]. - repository_url: str. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.agents_deployments_control_plane.validate_repository( repository_url="string", ) """ _request: typing.Dict[str, typing.Any] = {"repository_url": repository_url} if deployment_id is not OMIT: _request["deployment_id"] = deployment_id if pat is not OMIT: _request["pat"] = pat _response = self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", "api/v1beta1/deployments/validate-repository" ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), json=jsonable_encoder(_request), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(RepositoryValidationResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def get_deployment( self, deployment_id: str, *, include_events: typing.Optional[bool] = None, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> DeploymentResponse: """ Parameters: - deployment_id: str. - include_events: typing.Optional[bool]. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.agents_deployments_control_plane.get_deployment( deployment_id="string", ) """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1beta1/deployments/{deployment_id}"), params=remove_none_from_dict( {"include_events": include_events, "project_id": project_id, "organization_id": organization_id} ), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(DeploymentResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def delete_deployment( self, deployment_id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> typing.Any: """ Parameters: - deployment_id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.agents_deployments_control_plane.delete_deployment( deployment_id="string", ) """ _response = self._client_wrapper.httpx_client.request( "DELETE", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1beta1/deployments/{deployment_id}"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.Any, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def update_deployment( self, deployment_id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, deployment_file_path: typing.Optional[str] = OMIT, git_ref: typing.Optional[str] = OMIT, git_sha: typing.Optional[str] = OMIT, llama_deploy_version: typing.Optional[str] = OMIT, personal_access_token: typing.Optional[str] = OMIT, repo_url: typing.Optional[str] = OMIT, secrets: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT, static_assets_path: typing.Optional[str] = OMIT, ) -> DeploymentResponse: """ Update an existing deployment with patch-style changes Args: project_id: The project ID deployment_id: The deployment ID to update update_data: The patch-style update data Parameters: - deployment_id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. - deployment_file_path: typing.Optional[str]. - git_ref: typing.Optional[str]. - git_sha: typing.Optional[str]. - llama_deploy_version: typing.Optional[str]. - personal_access_token: typing.Optional[str]. - repo_url: typing.Optional[str]. - secrets: typing.Optional[typing.Dict[str, typing.Optional[str]]]. - static_assets_path: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.agents_deployments_control_plane.update_deployment( deployment_id="string", ) """ _request: typing.Dict[str, typing.Any] = {} if deployment_file_path is not OMIT: _request["deployment_file_path"] = deployment_file_path if git_ref is not OMIT: _request["git_ref"] = git_ref if git_sha is not OMIT: _request["git_sha"] = git_sha if llama_deploy_version is not OMIT: _request["llama_deploy_version"] = llama_deploy_version if personal_access_token is not OMIT: _request["personal_access_token"] = personal_access_token if repo_url is not OMIT: _request["repo_url"] = repo_url if secrets is not OMIT: _request["secrets"] = secrets if static_assets_path is not OMIT: _request["static_assets_path"] = static_assets_path _response = self._client_wrapper.httpx_client.request( "PATCH", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1beta1/deployments/{deployment_id}"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), json=jsonable_encoder(_request), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(DeploymentResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def get_deployment_history( self, deployment_id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> DeploymentHistoryResponse: """ Parameters: - deployment_id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.agents_deployments_control_plane.get_deployment_history( deployment_id="string", ) """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1beta1/deployments/{deployment_id}/history" ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(DeploymentHistoryResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def stream_deployment_logs( self, deployment_id: str, *, include_init_containers: typing.Optional[bool] = None, since_seconds: typing.Optional[int] = None, tail_lines: typing.Optional[int] = None, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> typing.Any: """ Stream logs for the latest ReplicaSet of a deployment. The stream ends when the latest ReplicaSet changes (e.g., a new rollout occurs). Parameters: - deployment_id: str. - include_init_containers: typing.Optional[bool]. - since_seconds: typing.Optional[int]. - tail_lines: typing.Optional[int]. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.agents_deployments_control_plane.stream_deployment_logs( deployment_id="string", ) """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1beta1/deployments/{deployment_id}/logs" ), params=remove_none_from_dict( { "include_init_containers": include_init_containers, "since_seconds": since_seconds, "tail_lines": tail_lines, "project_id": project_id, "organization_id": organization_id, } ), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.Any, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def rollback_deployment( self, deployment_id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, git_sha: str, ) -> DeploymentResponse: """ Parameters: - deployment_id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. - git_sha: str. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.agents_deployments_control_plane.rollback_deployment( deployment_id="string", git_sha="string", ) """ _response = self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1beta1/deployments/{deployment_id}/rollback" ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), json=jsonable_encoder({"git_sha": git_sha}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(DeploymentResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) class AsyncAgentsDeploymentsControlPlaneClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def get_deployments( self, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None ) -> DeploymentsListResponse: """ Parameters: - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.agents_deployments_control_plane.get_deployments() """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1beta1/deployments"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(DeploymentsListResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def create_deployment( self, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, deployment_file_path: typing.Optional[str] = OMIT, git_ref: typing.Optional[str] = OMIT, llama_deploy_version: typing.Optional[str] = OMIT, name: str, personal_access_token: typing.Optional[str] = OMIT, repo_url: str, secrets: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT, ) -> DeploymentResponse: """ Parameters: - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. - deployment_file_path: typing.Optional[str]. - git_ref: typing.Optional[str]. - llama_deploy_version: typing.Optional[str]. - name: str. - personal_access_token: typing.Optional[str]. - repo_url: str. - secrets: typing.Optional[typing.Dict[str, typing.Optional[str]]]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.agents_deployments_control_plane.create_deployment( name="string", repo_url="string", ) """ _request: typing.Dict[str, typing.Any] = {"name": name, "repo_url": repo_url} if deployment_file_path is not OMIT: _request["deployment_file_path"] = deployment_file_path if git_ref is not OMIT: _request["git_ref"] = git_ref if llama_deploy_version is not OMIT: _request["llama_deploy_version"] = llama_deploy_version if personal_access_token is not OMIT: _request["personal_access_token"] = personal_access_token if secrets is not OMIT: _request["secrets"] = secrets _response = await self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1beta1/deployments"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), json=jsonable_encoder(_request), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(DeploymentResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def get_version(self) -> VersionResponse: """ from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.agents_deployments_control_plane.get_version() """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1beta1/deployments-public/version"), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(VersionResponse, _response.json()) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def get_projects(self) -> ProjectsListResponse: """ Get all unique projects with their deployment counts --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.agents_deployments_control_plane.get_projects() """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1beta1/deployments/list-projects"), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(ProjectsListResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def validate_repository( self, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, deployment_id: typing.Optional[str] = OMIT, pat: typing.Optional[str] = OMIT, repository_url: str, ) -> RepositoryValidationResponse: """ Validate repository access and return unified response. Parameters: - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. - deployment_id: typing.Optional[str]. - pat: typing.Optional[str]. - repository_url: str. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.agents_deployments_control_plane.validate_repository( repository_url="string", ) """ _request: typing.Dict[str, typing.Any] = {"repository_url": repository_url} if deployment_id is not OMIT: _request["deployment_id"] = deployment_id if pat is not OMIT: _request["pat"] = pat _response = await self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", "api/v1beta1/deployments/validate-repository" ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), json=jsonable_encoder(_request), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(RepositoryValidationResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def get_deployment( self, deployment_id: str, *, include_events: typing.Optional[bool] = None, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> DeploymentResponse: """ Parameters: - deployment_id: str. - include_events: typing.Optional[bool]. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.agents_deployments_control_plane.get_deployment( deployment_id="string", ) """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1beta1/deployments/{deployment_id}"), params=remove_none_from_dict( {"include_events": include_events, "project_id": project_id, "organization_id": organization_id} ), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(DeploymentResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def delete_deployment( self, deployment_id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> typing.Any: """ Parameters: - deployment_id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.agents_deployments_control_plane.delete_deployment( deployment_id="string", ) """ _response = await self._client_wrapper.httpx_client.request( "DELETE", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1beta1/deployments/{deployment_id}"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.Any, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def update_deployment( self, deployment_id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, deployment_file_path: typing.Optional[str] = OMIT, git_ref: typing.Optional[str] = OMIT, git_sha: typing.Optional[str] = OMIT, llama_deploy_version: typing.Optional[str] = OMIT, personal_access_token: typing.Optional[str] = OMIT, repo_url: typing.Optional[str] = OMIT, secrets: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT, static_assets_path: typing.Optional[str] = OMIT, ) -> DeploymentResponse: """ Update an existing deployment with patch-style changes Args: project_id: The project ID deployment_id: The deployment ID to update update_data: The patch-style update data Parameters: - deployment_id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. - deployment_file_path: typing.Optional[str]. - git_ref: typing.Optional[str]. - git_sha: typing.Optional[str]. - llama_deploy_version: typing.Optional[str]. - personal_access_token: typing.Optional[str]. - repo_url: typing.Optional[str]. - secrets: typing.Optional[typing.Dict[str, typing.Optional[str]]]. - static_assets_path: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.agents_deployments_control_plane.update_deployment( deployment_id="string", ) """ _request: typing.Dict[str, typing.Any] = {} if deployment_file_path is not OMIT: _request["deployment_file_path"] = deployment_file_path if git_ref is not OMIT: _request["git_ref"] = git_ref if git_sha is not OMIT: _request["git_sha"] = git_sha if llama_deploy_version is not OMIT: _request["llama_deploy_version"] = llama_deploy_version if personal_access_token is not OMIT: _request["personal_access_token"] = personal_access_token if repo_url is not OMIT: _request["repo_url"] = repo_url if secrets is not OMIT: _request["secrets"] = secrets if static_assets_path is not OMIT: _request["static_assets_path"] = static_assets_path _response = await self._client_wrapper.httpx_client.request( "PATCH", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1beta1/deployments/{deployment_id}"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), json=jsonable_encoder(_request), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(DeploymentResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def get_deployment_history( self, deployment_id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> DeploymentHistoryResponse: """ Parameters: - deployment_id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.agents_deployments_control_plane.get_deployment_history( deployment_id="string", ) """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1beta1/deployments/{deployment_id}/history" ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(DeploymentHistoryResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def stream_deployment_logs( self, deployment_id: str, *, include_init_containers: typing.Optional[bool] = None, since_seconds: typing.Optional[int] = None, tail_lines: typing.Optional[int] = None, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> typing.Any: """ Stream logs for the latest ReplicaSet of a deployment. The stream ends when the latest ReplicaSet changes (e.g., a new rollout occurs). Parameters: - deployment_id: str. - include_init_containers: typing.Optional[bool]. - since_seconds: typing.Optional[int]. - tail_lines: typing.Optional[int]. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.agents_deployments_control_plane.stream_deployment_logs( deployment_id="string", ) """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1beta1/deployments/{deployment_id}/logs" ), params=remove_none_from_dict( { "include_init_containers": include_init_containers, "since_seconds": since_seconds, "tail_lines": tail_lines, "project_id": project_id, "organization_id": organization_id, } ), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.Any, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def rollback_deployment( self, deployment_id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, git_sha: str, ) -> DeploymentResponse: """ Parameters: - deployment_id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. - git_sha: str. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.agents_deployments_control_plane.rollback_deployment( deployment_id="string", git_sha="string", ) """ _response = await self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1beta1/deployments/{deployment_id}/rollback" ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), json=jsonable_encoder({"git_sha": git_sha}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(DeploymentResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json)