coolipy.resources.s3_storages

 1"""S3 storage 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.s3_storages import (
11    S3StorageCreateModel,
12    S3StorageModel,
13    S3StorageUpdateModel,
14)
15
16S3StorageList = list[S3StorageModel]
17
18
19def _dump(model: Any) -> dict[str, Any]:
20    return model.model_dump(mode="json", exclude_none=True)
21
22
23class S3Storages(ResourceBase):
24    """Synchronous client for Coolify S3 storages."""
25
26    def list(self) -> CoolipyAPIResponse[S3StorageList]:
27        """List all S3 storages."""
28        return self._get("/s3-storages", response_model=list[S3StorageModel])
29
30    def get(self, uuid: str) -> CoolipyAPIResponse[S3StorageModel]:
31        """Get an S3 storage by UUID."""
32        return self._get(f"/s3-storages/{uuid}", response_model=S3StorageModel)
33
34    def create(self, model: S3StorageCreateModel) -> CoolipyAPIResponse[UUIDResponse]:
35        """Create an S3 storage."""
36        return self._post("/s3-storages", json=_dump(model), response_model=UUIDResponse)
37
38    def update(self, uuid: str, model: S3StorageUpdateModel) -> CoolipyAPIResponse[S3StorageModel]:
39        """Update an S3 storage by UUID."""
40        return self._patch(f"/s3-storages/{uuid}", json=_dump(model), response_model=S3StorageModel)
41
42    def delete(self, uuid: str) -> CoolipyAPIResponse[MessageResponse]:
43        """Delete an S3 storage by UUID."""
44        return self._delete(f"/s3-storages/{uuid}", response_model=MessageResponse)
45
46    def validate(self, uuid: str) -> CoolipyAPIResponse[Any]:
47        """Validate an S3 storage by UUID."""
48        return self._post(f"/s3-storages/{uuid}/validate")
49
50
51class AsyncS3Storages(AsyncResourceBase):
52    """Asynchronous client for Coolify S3 storages."""
53
54    async def list(self) -> CoolipyAPIResponse[S3StorageList]:
55        """List all S3 storages."""
56        return await self._get("/s3-storages", response_model=list[S3StorageModel])
57
58    async def get(self, uuid: str) -> CoolipyAPIResponse[S3StorageModel]:
59        """Get an S3 storage by UUID."""
60        return await self._get(f"/s3-storages/{uuid}", response_model=S3StorageModel)
61
62    async def create(self, model: S3StorageCreateModel) -> CoolipyAPIResponse[UUIDResponse]:
63        """Create an S3 storage."""
64        return await self._post("/s3-storages", json=_dump(model), response_model=UUIDResponse)
65
66    async def update(
67        self, uuid: str, model: S3StorageUpdateModel
68    ) -> CoolipyAPIResponse[S3StorageModel]:
69        """Update an S3 storage by UUID."""
70        return await self._patch(
71            f"/s3-storages/{uuid}", json=_dump(model), response_model=S3StorageModel
72        )
73
74    async def delete(self, uuid: str) -> CoolipyAPIResponse[MessageResponse]:
75        """Delete an S3 storage by UUID."""
76        return await self._delete(f"/s3-storages/{uuid}", response_model=MessageResponse)
77
78    async def validate(self, uuid: str) -> CoolipyAPIResponse[Any]:
79        """Validate an S3 storage by UUID."""
80        return await self._post(f"/s3-storages/{uuid}/validate")