coolipy.resources.servers

  1"""Servers resource clients (sync + async)."""
  2
  3from __future__ import annotations
  4
  5from typing import Any
  6
  7from coolipy._base import AsyncResourceBase, ResourceBase
  8from coolipy._response import CoolipyAPIResponse
  9from coolipy.models.common import (
 10    Destination,
 11    DestinationCreate,
 12    EnvironmentVariable,
 13    EnvironmentVariableCreate,
 14    MessageResponse,
 15    UUIDResponse,
 16)
 17from coolipy.models.servers import ServerCreateModel, ServerModel, ServerUpdateModel
 18
 19ServerList = list[ServerModel]
 20DestinationList = list[Destination]
 21EnvironmentVariableList = list[EnvironmentVariable]
 22
 23
 24def _dump(model: Any) -> dict[str, Any]:
 25    return model.model_dump(mode="json", exclude_none=True)
 26
 27
 28class Servers(ResourceBase):
 29    """Synchronous client for Coolify servers."""
 30
 31    def list(self) -> CoolipyAPIResponse[ServerList]:
 32        """List all servers."""
 33        return self._get("/servers", response_model=list[ServerModel])
 34
 35    def get(self, uuid: str) -> CoolipyAPIResponse[ServerModel]:
 36        """Get a server by UUID."""
 37        return self._get(f"/servers/{uuid}", response_model=ServerModel)
 38
 39    def create(self, model: ServerCreateModel) -> CoolipyAPIResponse[UUIDResponse]:
 40        """Create a server."""
 41        return self._post("/servers", json=_dump(model), response_model=UUIDResponse)
 42
 43    def update(self, uuid: str, model: ServerUpdateModel) -> CoolipyAPIResponse[ServerModel]:
 44        """Update a server by UUID."""
 45        return self._patch(f"/servers/{uuid}", json=_dump(model), response_model=ServerModel)
 46
 47    def delete(self, uuid: str) -> CoolipyAPIResponse[MessageResponse]:
 48        """Delete a server by UUID."""
 49        return self._delete(f"/servers/{uuid}", response_model=MessageResponse)
 50
 51    def resources(self, uuid: str) -> CoolipyAPIResponse[Any]:
 52        """List resources running on a server."""
 53        return self._get(f"/servers/{uuid}/resources")
 54
 55    def domains(self, uuid: str) -> CoolipyAPIResponse[Any]:
 56        """List domains configured on a server."""
 57        return self._get(f"/servers/{uuid}/domains")
 58
 59    def validate(self, uuid: str) -> CoolipyAPIResponse[Any]:
 60        """Validate a server."""
 61        return self._post(f"/servers/{uuid}/validate")
 62
 63    def destinations(self, server_uuid: str) -> CoolipyAPIResponse[DestinationList]:
 64        """List destinations for a server."""
 65        return self._get(f"/servers/{server_uuid}/destinations", response_model=list[Destination])
 66
 67    def add_destination(
 68        self, server_uuid: str, model: DestinationCreate
 69    ) -> CoolipyAPIResponse[Any]:
 70        """Add a destination to a server."""
 71        return self._post(f"/servers/{server_uuid}/destinations", json=_dump(model))
 72
 73    def envs(self, uuid: str) -> CoolipyAPIResponse[EnvironmentVariableList]:
 74        """List environment variables for a server."""
 75        return self._get(f"/servers/{uuid}/envs", response_model=list[EnvironmentVariable])
 76
 77    def create_env(
 78        self, uuid: str, model: EnvironmentVariableCreate
 79    ) -> CoolipyAPIResponse[UUIDResponse]:
 80        """Create an environment variable for a server."""
 81        return self._post(f"/servers/{uuid}/envs", json=_dump(model), response_model=UUIDResponse)
 82
 83    def delete_env(self, uuid: str, env_id: str) -> CoolipyAPIResponse[MessageResponse]:
 84        """Delete an environment variable from a server."""
 85        return self._delete(f"/servers/{uuid}/envs/{env_id}", response_model=MessageResponse)
 86
 87
 88class AsyncServers(AsyncResourceBase):
 89    """Asynchronous client for Coolify servers."""
 90
 91    async def list(self) -> CoolipyAPIResponse[ServerList]:
 92        """List all servers."""
 93        return await self._get("/servers", response_model=list[ServerModel])
 94
 95    async def get(self, uuid: str) -> CoolipyAPIResponse[ServerModel]:
 96        """Get a server by UUID."""
 97        return await self._get(f"/servers/{uuid}", response_model=ServerModel)
 98
 99    async def create(self, model: ServerCreateModel) -> CoolipyAPIResponse[UUIDResponse]:
100        """Create a server."""
101        return await self._post("/servers", json=_dump(model), response_model=UUIDResponse)
102
103    async def update(self, uuid: str, model: ServerUpdateModel) -> CoolipyAPIResponse[ServerModel]:
104        """Update a server by UUID."""
105        return await self._patch(f"/servers/{uuid}", json=_dump(model), response_model=ServerModel)
106
107    async def delete(self, uuid: str) -> CoolipyAPIResponse[MessageResponse]:
108        """Delete a server by UUID."""
109        return await self._delete(f"/servers/{uuid}", response_model=MessageResponse)
110
111    async def resources(self, uuid: str) -> CoolipyAPIResponse[Any]:
112        """List resources running on a server."""
113        return await self._get(f"/servers/{uuid}/resources")
114
115    async def domains(self, uuid: str) -> CoolipyAPIResponse[Any]:
116        """List domains configured on a server."""
117        return await self._get(f"/servers/{uuid}/domains")
118
119    async def validate(self, uuid: str) -> CoolipyAPIResponse[Any]:
120        """Validate a server."""
121        return await self._post(f"/servers/{uuid}/validate")
122
123    async def destinations(self, server_uuid: str) -> CoolipyAPIResponse[DestinationList]:
124        """List destinations for a server."""
125        return await self._get(
126            f"/servers/{server_uuid}/destinations", response_model=list[Destination]
127        )
128
129    async def add_destination(
130        self, server_uuid: str, model: DestinationCreate
131    ) -> CoolipyAPIResponse[Any]:
132        """Add a destination to a server."""
133        return await self._post(f"/servers/{server_uuid}/destinations", json=_dump(model))
134
135    async def envs(self, uuid: str) -> CoolipyAPIResponse[EnvironmentVariableList]:
136        """List environment variables for a server."""
137        return await self._get(f"/servers/{uuid}/envs", response_model=list[EnvironmentVariable])
138
139    async def create_env(
140        self, uuid: str, model: EnvironmentVariableCreate
141    ) -> CoolipyAPIResponse[UUIDResponse]:
142        """Create an environment variable for a server."""
143        return await self._post(
144            f"/servers/{uuid}/envs", json=_dump(model), response_model=UUIDResponse
145        )
146
147    async def delete_env(self, uuid: str, env_id: str) -> CoolipyAPIResponse[MessageResponse]:
148        """Delete an environment variable from a server."""
149        return await self._delete(f"/servers/{uuid}/envs/{env_id}", response_model=MessageResponse)