coolipy.resources.projects

  1"""Projects 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 MessageResponse, UUIDResponse
 10from coolipy.models.projects import (
 11    EnvironmentCreateModel,
 12    EnvironmentModel,
 13    EnvironmentUpdateModel,
 14    ProjectCreateModel,
 15    ProjectModel,
 16    ProjectUpdateModel,
 17)
 18
 19ProjectList = list[ProjectModel]
 20EnvironmentList = list[EnvironmentModel]
 21
 22
 23def _dump(model: Any) -> dict[str, Any]:
 24    return model.model_dump(mode="json", exclude_none=True)
 25
 26
 27class Projects(ResourceBase):
 28    """Synchronous client for Coolify projects."""
 29
 30    def list(self) -> CoolipyAPIResponse[ProjectList]:
 31        """List all projects."""
 32        return self._get("/projects", response_model=list[ProjectModel])
 33
 34    def get(self, uuid: str) -> CoolipyAPIResponse[ProjectModel]:
 35        """Get a project by UUID."""
 36        return self._get(f"/projects/{uuid}", response_model=ProjectModel)
 37
 38    def create(self, model: ProjectCreateModel) -> CoolipyAPIResponse[UUIDResponse]:
 39        """Create a project."""
 40        return self._post("/projects", json=_dump(model), response_model=UUIDResponse)
 41
 42    def update(self, uuid: str, model: ProjectUpdateModel) -> CoolipyAPIResponse[dict]:
 43        """Update a project by UUID."""
 44        return self._patch(f"/projects/{uuid}", json=_dump(model), response_model=dict)
 45
 46    def delete(self, uuid: str) -> CoolipyAPIResponse[MessageResponse]:
 47        """Delete a project by UUID."""
 48        return self._delete(f"/projects/{uuid}", response_model=MessageResponse)
 49
 50    def environment(
 51        self, uuid: str, environment_name_or_uuid: str
 52    ) -> CoolipyAPIResponse[EnvironmentModel]:
 53        """Get an environment by name or UUID."""
 54        return self._get(
 55            f"/projects/{uuid}/{environment_name_or_uuid}", response_model=EnvironmentModel
 56        )
 57
 58    def environments(self, uuid: str) -> CoolipyAPIResponse[EnvironmentList]:
 59        """List environments for a project."""
 60        return self._get(f"/projects/{uuid}/environments", response_model=list[EnvironmentModel])
 61
 62    def create_environment(
 63        self, uuid: str, model: EnvironmentCreateModel
 64    ) -> CoolipyAPIResponse[UUIDResponse]:
 65        """Create an environment for a project."""
 66        return self._post(
 67            f"/projects/{uuid}/environments", json=_dump(model), response_model=UUIDResponse
 68        )
 69
 70    def update_environment(
 71        self, uuid: str, environment_name_or_uuid: str, model: EnvironmentUpdateModel
 72    ) -> CoolipyAPIResponse[EnvironmentModel]:
 73        """Update an environment by name or UUID."""
 74        return self._patch(
 75            f"/projects/{uuid}/environments/{environment_name_or_uuid}",
 76            json=_dump(model),
 77            response_model=EnvironmentModel,
 78        )
 79
 80    def delete_environment(
 81        self, uuid: str, environment_name_or_uuid: str
 82    ) -> CoolipyAPIResponse[MessageResponse]:
 83        """Delete an environment by name or UUID."""
 84        return self._delete(
 85            f"/projects/{uuid}/environments/{environment_name_or_uuid}",
 86            response_model=MessageResponse,
 87        )
 88
 89
 90class AsyncProjects(AsyncResourceBase):
 91    """Asynchronous client for Coolify projects."""
 92
 93    async def list(self) -> CoolipyAPIResponse[ProjectList]:
 94        """List all projects."""
 95        return await self._get("/projects", response_model=list[ProjectModel])
 96
 97    async def get(self, uuid: str) -> CoolipyAPIResponse[ProjectModel]:
 98        """Get a project by UUID."""
 99        return await self._get(f"/projects/{uuid}", response_model=ProjectModel)
100
101    async def create(self, model: ProjectCreateModel) -> CoolipyAPIResponse[UUIDResponse]:
102        """Create a project."""
103        return await self._post("/projects", json=_dump(model), response_model=UUIDResponse)
104
105    async def update(self, uuid: str, model: ProjectUpdateModel) -> CoolipyAPIResponse[dict]:
106        """Update a project by UUID."""
107        return await self._patch(f"/projects/{uuid}", json=_dump(model), response_model=dict)
108
109    async def delete(self, uuid: str) -> CoolipyAPIResponse[MessageResponse]:
110        """Delete a project by UUID."""
111        return await self._delete(f"/projects/{uuid}", response_model=MessageResponse)
112
113    async def environment(
114        self, uuid: str, environment_name_or_uuid: str
115    ) -> CoolipyAPIResponse[EnvironmentModel]:
116        """Get an environment by name or UUID."""
117        return await self._get(
118            f"/projects/{uuid}/{environment_name_or_uuid}", response_model=EnvironmentModel
119        )
120
121    async def environments(self, uuid: str) -> CoolipyAPIResponse[EnvironmentList]:
122        """List environments for a project."""
123        return await self._get(
124            f"/projects/{uuid}/environments", response_model=list[EnvironmentModel]
125        )
126
127    async def create_environment(
128        self, uuid: str, model: EnvironmentCreateModel
129    ) -> CoolipyAPIResponse[UUIDResponse]:
130        """Create an environment for a project."""
131        return await self._post(
132            f"/projects/{uuid}/environments", json=_dump(model), response_model=UUIDResponse
133        )
134
135    async def update_environment(
136        self, uuid: str, environment_name_or_uuid: str, model: EnvironmentUpdateModel
137    ) -> CoolipyAPIResponse[EnvironmentModel]:
138        """Update an environment by name or UUID."""
139        return await self._patch(
140            f"/projects/{uuid}/environments/{environment_name_or_uuid}",
141            json=_dump(model),
142            response_model=EnvironmentModel,
143        )
144
145    async def delete_environment(
146        self, uuid: str, environment_name_or_uuid: str
147    ) -> CoolipyAPIResponse[MessageResponse]:
148        """Delete an environment by name or UUID."""
149        return await self._delete(
150            f"/projects/{uuid}/environments/{environment_name_or_uuid}",
151            response_model=MessageResponse,
152        )