coolipy.resources.deployments

  1"""Deployments resource clients (sync + async)."""
  2
  3from __future__ import annotations
  4
  5from coolipy._base import AsyncResourceBase, ResourceBase
  6from coolipy._response import CoolipyAPIResponse
  7from coolipy.models.applications import ApplicationModel
  8from coolipy.models.deployments import DeploymentModel, DeployResponse
  9
 10DeploymentList = list[DeploymentModel]
 11ApplicationList = list[ApplicationModel]
 12
 13
 14class Deployments(ResourceBase):
 15    """Synchronous client for Coolify deployments."""
 16
 17    def list(self) -> CoolipyAPIResponse[DeploymentList]:
 18        """List currently running deployments."""
 19        return self._get("/deployments", response_model=list[DeploymentModel])
 20
 21    def get(self, uuid: str) -> CoolipyAPIResponse[DeploymentModel]:
 22        """Get a deployment by UUID."""
 23        return self._get(f"/deployments/{uuid}", response_model=DeploymentModel)
 24
 25    def cancel(self, uuid: str) -> CoolipyAPIResponse[dict]:
 26        """Cancel a deployment by UUID."""
 27        return self._post(f"/deployments/{uuid}/cancel", response_model=dict)
 28
 29    def deploy(
 30        self,
 31        *,
 32        tag: str | None = None,
 33        uuid: str | None = None,
 34        force: bool | None = None,
 35        pr: int | None = None,
 36        pull_request_id: int | None = None,
 37        docker_tag: str | None = None,
 38    ) -> CoolipyAPIResponse[DeployResponse]:
 39        """Deploy resources by tag or UUID."""
 40        params: dict[str, object] = {}
 41        if tag is not None:
 42            params["tag"] = tag
 43        if uuid is not None:
 44            params["uuid"] = uuid
 45        if force is not None:
 46            params["force"] = force
 47        if pr is not None:
 48            params["pr"] = pr
 49        if pull_request_id is not None:
 50            params["pull_request_id"] = pull_request_id
 51        if docker_tag is not None:
 52            params["docker_tag"] = docker_tag
 53        return self._post("/deploy", params=params, response_model=DeployResponse)
 54
 55    def application_deployments(
 56        self, uuid: str, *, skip: int = 0, take: int = 10
 57    ) -> CoolipyAPIResponse[ApplicationList]:
 58        """List deployments for an application."""
 59        return self._get(
 60            f"/deployments/applications/{uuid}",
 61            params={"skip": skip, "take": take},
 62            response_model=list[ApplicationModel],
 63        )
 64
 65
 66class AsyncDeployments(AsyncResourceBase):
 67    """Asynchronous client for Coolify deployments."""
 68
 69    async def list(self) -> CoolipyAPIResponse[DeploymentList]:
 70        """List currently running deployments."""
 71        return await self._get("/deployments", response_model=list[DeploymentModel])
 72
 73    async def get(self, uuid: str) -> CoolipyAPIResponse[DeploymentModel]:
 74        """Get a deployment by UUID."""
 75        return await self._get(f"/deployments/{uuid}", response_model=DeploymentModel)
 76
 77    async def cancel(self, uuid: str) -> CoolipyAPIResponse[dict]:
 78        """Cancel a deployment by UUID."""
 79        return await self._post(f"/deployments/{uuid}/cancel", response_model=dict)
 80
 81    async def deploy(
 82        self,
 83        *,
 84        tag: str | None = None,
 85        uuid: str | None = None,
 86        force: bool | None = None,
 87        pr: int | None = None,
 88        pull_request_id: int | None = None,
 89        docker_tag: str | None = None,
 90    ) -> CoolipyAPIResponse[DeployResponse]:
 91        """Deploy resources by tag or UUID."""
 92        params: dict[str, object] = {}
 93        if tag is not None:
 94            params["tag"] = tag
 95        if uuid is not None:
 96            params["uuid"] = uuid
 97        if force is not None:
 98            params["force"] = force
 99        if pr is not None:
100            params["pr"] = pr
101        if pull_request_id is not None:
102            params["pull_request_id"] = pull_request_id
103        if docker_tag is not None:
104            params["docker_tag"] = docker_tag
105        return await self._post("/deploy", params=params, response_model=DeployResponse)
106
107    async def application_deployments(
108        self, uuid: str, *, skip: int = 0, take: int = 10
109    ) -> CoolipyAPIResponse[ApplicationList]:
110        """List deployments for an application."""
111        return await self._get(
112            f"/deployments/applications/{uuid}",
113            params={"skip": skip, "take": take},
114            response_model=list[ApplicationModel],
115        )