coolipy.resources.databases

  1"""Databases 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.exceptions import CoolipyValidationError
 10from coolipy.models.common import (
 11    BulkEnvsUpdate,
 12    DeploymentQueuedResponse,
 13    EnvironmentVariable,
 14    EnvironmentVariableCreate,
 15    EnvironmentVariableUpdate,
 16    Logs,
 17    MessageResponse,
 18    StorageCreate,
 19    StorageUpdate,
 20    Tag,
 21    TagsCreate,
 22    UUIDResponse,
 23    VolumeBackupScheduleRequest,
 24    VolumeBackupScheduleResponse,
 25)
 26from coolipy.models.databases import (
 27    ClickhouseModelCreate,
 28    DatabaseBackupCreate,
 29    DatabaseModel,
 30    DatabaseUpdateModel,
 31    DragonflyModelCreate,
 32    KeyDBModelCreate,
 33    MariaDBModelCreate,
 34    MongoDBModelCreate,
 35    MySQLModelCreate,
 36    PostgreSQLModelCreate,
 37    RedisModelCreate,
 38)
 39
 40DatabaseList = list[DatabaseModel]
 41EnvironmentVariableList = list[EnvironmentVariable]
 42TagList = list[Tag]
 43
 44_DATABASE_CREATE_PATHS: dict[type, str] = {
 45    PostgreSQLModelCreate: "/databases/postgresql",
 46    MySQLModelCreate: "/databases/mysql",
 47    MariaDBModelCreate: "/databases/mariadb",
 48    MongoDBModelCreate: "/databases/mongodb",
 49    RedisModelCreate: "/databases/redis",
 50    ClickhouseModelCreate: "/databases/clickhouse",
 51    DragonflyModelCreate: "/databases/dragonfly",
 52    KeyDBModelCreate: "/databases/keydb",
 53}
 54
 55
 56def _dump(model: Any) -> dict[str, Any]:
 57    return model.model_dump(mode="json", exclude_none=True)
 58
 59
 60class Databases(ResourceBase):
 61    """Synchronous client for Coolify databases."""
 62
 63    def list(self) -> CoolipyAPIResponse[DatabaseList]:
 64        """List all databases."""
 65        return self._get("/databases", response_model=list[DatabaseModel])
 66
 67    def get(self, uuid: str) -> CoolipyAPIResponse[DatabaseModel]:
 68        """Get a database by UUID."""
 69        return self._get(f"/databases/{uuid}", response_model=DatabaseModel)
 70
 71    def create(self, model: Any) -> CoolipyAPIResponse[Any]:
 72        """Create a database from one of the supported create models."""
 73        path = _DATABASE_CREATE_PATHS.get(type(model))
 74        if path is None:
 75            raise CoolipyValidationError(
 76                f"Unsupported database create model: {type(model).__name__}"
 77            )
 78        return self._post(path, json=_dump(model))
 79
 80    def update(self, uuid: str, model: DatabaseUpdateModel) -> CoolipyAPIResponse[Any]:
 81        """Update a database by UUID."""
 82        return self._patch(f"/databases/{uuid}", json=_dump(model))
 83
 84    def delete(
 85        self,
 86        uuid: str,
 87        *,
 88        delete_configurations: bool = True,
 89        delete_volumes: bool = True,
 90        docker_cleanup: bool = True,
 91        delete_connected_networks: bool = True,
 92    ) -> CoolipyAPIResponse[MessageResponse]:
 93        """Delete a database by UUID."""
 94        params = {
 95            "delete_configurations": delete_configurations,
 96            "delete_volumes": delete_volumes,
 97            "docker_cleanup": docker_cleanup,
 98            "delete_connected_networks": delete_connected_networks,
 99        }
100        return self._delete(f"/databases/{uuid}", params=params, response_model=MessageResponse)
101
102    def backups(self, uuid: str) -> CoolipyAPIResponse[Any]:
103        """List scheduled backups for a database."""
104        return self._get(f"/databases/{uuid}/backups")
105
106    def create_backup(self, uuid: str, model: DatabaseBackupCreate) -> CoolipyAPIResponse[dict]:
107        """Create a scheduled backup configuration for a database."""
108        return self._post(f"/databases/{uuid}/backups", json=_dump(model), response_model=dict)
109
110    def update_backup(
111        self, uuid: str, scheduled_backup_uuid: str, model: DatabaseBackupCreate
112    ) -> CoolipyAPIResponse[Any]:
113        """Update a scheduled backup configuration."""
114        return self._patch(f"/databases/{uuid}/backups/{scheduled_backup_uuid}", json=_dump(model))
115
116    def delete_backup(self, uuid: str, scheduled_backup_uuid: str) -> CoolipyAPIResponse[Any]:
117        """Delete a scheduled backup configuration."""
118        return self._delete(f"/databases/{uuid}/backups/{scheduled_backup_uuid}")
119
120    def backup_executions(self, uuid: str, scheduled_backup_uuid: str) -> CoolipyAPIResponse[Any]:
121        """List executions of a scheduled backup."""
122        return self._get(f"/databases/{uuid}/backups/{scheduled_backup_uuid}/executions")
123
124    def delete_backup_execution(
125        self, uuid: str, scheduled_backup_uuid: str, execution_uuid: str
126    ) -> CoolipyAPIResponse[Any]:
127        """Delete a backup execution."""
128        return self._delete(
129            f"/databases/{uuid}/backups/{scheduled_backup_uuid}/executions/{execution_uuid}"
130        )
131
132    def logs(self, uuid: str) -> CoolipyAPIResponse[Logs]:
133        """Get database logs."""
134        return self._get(f"/databases/{uuid}/logs", response_model=Logs)
135
136    def move(self, uuid: str, environment_uuid: str) -> CoolipyAPIResponse[dict]:
137        """Move a database to another environment."""
138        return self._post(
139            f"/databases/{uuid}/move",
140            json={"environment_uuid": environment_uuid},
141            response_model=dict,
142        )
143
144    def migrate(
145        self, uuid: str, destination_uuid: str, *, migrate_volumes: bool = True
146    ) -> CoolipyAPIResponse[Any]:
147        """Migrate a database to another destination/server."""
148        body = {"destination_uuid": destination_uuid, "migrate_volumes": migrate_volumes}
149        return self._post(f"/databases/{uuid}/migrate", json=body)
150
151    def start(self, uuid: str) -> CoolipyAPIResponse[DeploymentQueuedResponse]:
152        """Start a database."""
153        return self._post(f"/databases/{uuid}/start", response_model=DeploymentQueuedResponse)
154
155    def stop(self, uuid: str) -> CoolipyAPIResponse[MessageResponse]:
156        """Stop a database."""
157        return self._post(f"/databases/{uuid}/stop", response_model=MessageResponse)
158
159    def restart(self, uuid: str) -> CoolipyAPIResponse[DeploymentQueuedResponse]:
160        """Restart a database."""
161        return self._post(f"/databases/{uuid}/restart", response_model=DeploymentQueuedResponse)
162
163    def envs(self, uuid: str) -> CoolipyAPIResponse[EnvironmentVariableList]:
164        """List environment variables for a database."""
165        return self._get(f"/databases/{uuid}/envs", response_model=list[EnvironmentVariable])
166
167    def create_env(
168        self, uuid: str, model: EnvironmentVariableCreate
169    ) -> CoolipyAPIResponse[UUIDResponse]:
170        """Create an environment variable for a database."""
171        return self._post(f"/databases/{uuid}/envs", json=_dump(model), response_model=UUIDResponse)
172
173    def update_env(
174        self, uuid: str, model: EnvironmentVariableUpdate
175    ) -> CoolipyAPIResponse[EnvironmentVariable]:
176        """Update an environment variable for a database."""
177        return self._patch(
178            f"/databases/{uuid}/envs", json=_dump(model), response_model=EnvironmentVariable
179        )
180
181    def bulk_update_envs(
182        self, uuid: str, model: BulkEnvsUpdate
183    ) -> CoolipyAPIResponse[EnvironmentVariableList]:
184        """Bulk-update environment variables for a database."""
185        return self._patch(
186            f"/databases/{uuid}/envs/bulk",
187            json=_dump(model),
188            response_model=list[EnvironmentVariable],
189        )
190
191    def delete_env(self, uuid: str, env_uuid: str) -> CoolipyAPIResponse[MessageResponse]:
192        """Delete an environment variable by UUID."""
193        return self._delete(f"/databases/{uuid}/envs/{env_uuid}", response_model=MessageResponse)
194
195    def storages(self, uuid: str) -> CoolipyAPIResponse[dict]:
196        """List storages for a database."""
197        return self._get(f"/databases/{uuid}/storages", response_model=dict)
198
199    def create_storage(self, uuid: str, model: StorageCreate) -> CoolipyAPIResponse[dict]:
200        """Create a storage for a database."""
201        return self._post(f"/databases/{uuid}/storages", json=_dump(model), response_model=dict)
202
203    def update_storage(self, uuid: str, model: StorageUpdate) -> CoolipyAPIResponse[dict]:
204        """Update a storage for a database."""
205        return self._patch(f"/databases/{uuid}/storages", json=_dump(model), response_model=dict)
206
207    def delete_storage(self, uuid: str, storage_uuid: str) -> CoolipyAPIResponse[MessageResponse]:
208        """Delete a storage by UUID."""
209        return self._delete(
210            f"/databases/{uuid}/storages/{storage_uuid}", response_model=MessageResponse
211        )
212
213    def tags(self, uuid: str) -> CoolipyAPIResponse[TagList]:
214        """List tags for a database."""
215        return self._get(f"/databases/{uuid}/tags", response_model=list[Tag])
216
217    def add_tags(self, uuid: str, model: TagsCreate) -> CoolipyAPIResponse[TagList]:
218        """Add one or more tags to a database."""
219        return self._post(f"/databases/{uuid}/tags", json=_dump(model), response_model=list[Tag])
220
221    def delete_tag(self, uuid: str, tag_uuid: str) -> CoolipyAPIResponse[Any]:
222        """Remove a tag from a database."""
223        return self._delete(f"/databases/{uuid}/tags/{tag_uuid}")
224
225    def clone(
226        self,
227        uuid: str,
228        destination_uuid: str,
229        *,
230        name: str | None = None,
231        clone_volumes: bool = False,
232    ) -> CoolipyAPIResponse[dict]:
233        """Clone a database into a destination."""
234        body = {"destination_uuid": destination_uuid, "name": name, "clone_volumes": clone_volumes}
235        return self._post(f"/databases/{uuid}/clone", json=body, response_model=dict)
236
237    def update_storage_backup(
238        self, uuid: str, storage_uuid: str, model: VolumeBackupScheduleRequest
239    ) -> CoolipyAPIResponse[VolumeBackupScheduleResponse]:
240        """Schedule backups for a storage volume."""
241        return self._put(
242            f"/databases/{uuid}/storages/{storage_uuid}/backups",
243            json=_dump(model),
244            response_model=VolumeBackupScheduleResponse,
245        )
246
247    def delete_storage_backup(
248        self, uuid: str, storage_uuid: str
249    ) -> CoolipyAPIResponse[MessageResponse]:
250        """Remove the backup schedule for a storage volume."""
251        return self._delete(
252            f"/databases/{uuid}/storages/{storage_uuid}/backups", response_model=MessageResponse
253        )
254
255    def run_storage_backup(
256        self, uuid: str, storage_uuid: str
257    ) -> CoolipyAPIResponse[MessageResponse]:
258        """Run a storage backup now."""
259        return self._post(
260            f"/databases/{uuid}/storages/{storage_uuid}/backups/run", response_model=MessageResponse
261        )
262
263
264class AsyncDatabases(AsyncResourceBase):
265    """Asynchronous client for Coolify databases."""
266
267    async def list(self) -> CoolipyAPIResponse[DatabaseList]:
268        """List all databases."""
269        return await self._get("/databases", response_model=list[DatabaseModel])
270
271    async def get(self, uuid: str) -> CoolipyAPIResponse[DatabaseModel]:
272        """Get a database by UUID."""
273        return await self._get(f"/databases/{uuid}", response_model=DatabaseModel)
274
275    async def create(self, model: Any) -> CoolipyAPIResponse[Any]:
276        """Create a database from one of the supported create models."""
277        path = _DATABASE_CREATE_PATHS.get(type(model))
278        if path is None:
279            raise CoolipyValidationError(
280                f"Unsupported database create model: {type(model).__name__}"
281            )
282        return await self._post(path, json=_dump(model))
283
284    async def update(self, uuid: str, model: DatabaseUpdateModel) -> CoolipyAPIResponse[Any]:
285        """Update a database by UUID."""
286        return await self._patch(f"/databases/{uuid}", json=_dump(model))
287
288    async def delete(
289        self,
290        uuid: str,
291        *,
292        delete_configurations: bool = True,
293        delete_volumes: bool = True,
294        docker_cleanup: bool = True,
295        delete_connected_networks: bool = True,
296    ) -> CoolipyAPIResponse[MessageResponse]:
297        """Delete a database by UUID."""
298        params = {
299            "delete_configurations": delete_configurations,
300            "delete_volumes": delete_volumes,
301            "docker_cleanup": docker_cleanup,
302            "delete_connected_networks": delete_connected_networks,
303        }
304        return await self._delete(
305            f"/databases/{uuid}", params=params, response_model=MessageResponse
306        )
307
308    async def backups(self, uuid: str) -> CoolipyAPIResponse[Any]:
309        """List scheduled backups for a database."""
310        return await self._get(f"/databases/{uuid}/backups")
311
312    async def create_backup(
313        self, uuid: str, model: DatabaseBackupCreate
314    ) -> CoolipyAPIResponse[dict]:
315        """Create a scheduled backup configuration for a database."""
316        return await self._post(
317            f"/databases/{uuid}/backups", json=_dump(model), response_model=dict
318        )
319
320    async def update_backup(
321        self, uuid: str, scheduled_backup_uuid: str, model: DatabaseBackupCreate
322    ) -> CoolipyAPIResponse[Any]:
323        """Update a scheduled backup configuration."""
324        return await self._patch(
325            f"/databases/{uuid}/backups/{scheduled_backup_uuid}", json=_dump(model)
326        )
327
328    async def delete_backup(self, uuid: str, scheduled_backup_uuid: str) -> CoolipyAPIResponse[Any]:
329        """Delete a scheduled backup configuration."""
330        return await self._delete(f"/databases/{uuid}/backups/{scheduled_backup_uuid}")
331
332    async def backup_executions(
333        self, uuid: str, scheduled_backup_uuid: str
334    ) -> CoolipyAPIResponse[Any]:
335        """List executions of a scheduled backup."""
336        return await self._get(f"/databases/{uuid}/backups/{scheduled_backup_uuid}/executions")
337
338    async def delete_backup_execution(
339        self, uuid: str, scheduled_backup_uuid: str, execution_uuid: str
340    ) -> CoolipyAPIResponse[Any]:
341        """Delete a backup execution."""
342        return await self._delete(
343            f"/databases/{uuid}/backups/{scheduled_backup_uuid}/executions/{execution_uuid}"
344        )
345
346    async def logs(self, uuid: str) -> CoolipyAPIResponse[Logs]:
347        """Get database logs."""
348        return await self._get(f"/databases/{uuid}/logs", response_model=Logs)
349
350    async def move(self, uuid: str, environment_uuid: str) -> CoolipyAPIResponse[dict]:
351        """Move a database to another environment."""
352        return await self._post(
353            f"/databases/{uuid}/move",
354            json={"environment_uuid": environment_uuid},
355            response_model=dict,
356        )
357
358    async def migrate(
359        self, uuid: str, destination_uuid: str, *, migrate_volumes: bool = True
360    ) -> CoolipyAPIResponse[Any]:
361        """Migrate a database to another destination/server."""
362        body = {"destination_uuid": destination_uuid, "migrate_volumes": migrate_volumes}
363        return await self._post(f"/databases/{uuid}/migrate", json=body)
364
365    async def start(self, uuid: str) -> CoolipyAPIResponse[DeploymentQueuedResponse]:
366        """Start a database."""
367        return await self._post(f"/databases/{uuid}/start", response_model=DeploymentQueuedResponse)
368
369    async def stop(self, uuid: str) -> CoolipyAPIResponse[MessageResponse]:
370        """Stop a database."""
371        return await self._post(f"/databases/{uuid}/stop", response_model=MessageResponse)
372
373    async def restart(self, uuid: str) -> CoolipyAPIResponse[DeploymentQueuedResponse]:
374        """Restart a database."""
375        return await self._post(
376            f"/databases/{uuid}/restart", response_model=DeploymentQueuedResponse
377        )
378
379    async def envs(self, uuid: str) -> CoolipyAPIResponse[EnvironmentVariableList]:
380        """List environment variables for a database."""
381        return await self._get(f"/databases/{uuid}/envs", response_model=list[EnvironmentVariable])
382
383    async def create_env(
384        self, uuid: str, model: EnvironmentVariableCreate
385    ) -> CoolipyAPIResponse[UUIDResponse]:
386        """Create an environment variable for a database."""
387        return await self._post(
388            f"/databases/{uuid}/envs", json=_dump(model), response_model=UUIDResponse
389        )
390
391    async def update_env(
392        self, uuid: str, model: EnvironmentVariableUpdate
393    ) -> CoolipyAPIResponse[EnvironmentVariable]:
394        """Update an environment variable for a database."""
395        return await self._patch(
396            f"/databases/{uuid}/envs", json=_dump(model), response_model=EnvironmentVariable
397        )
398
399    async def bulk_update_envs(
400        self, uuid: str, model: BulkEnvsUpdate
401    ) -> CoolipyAPIResponse[EnvironmentVariableList]:
402        """Bulk-update environment variables for a database."""
403        return await self._patch(
404            f"/databases/{uuid}/envs/bulk",
405            json=_dump(model),
406            response_model=list[EnvironmentVariable],
407        )
408
409    async def delete_env(self, uuid: str, env_uuid: str) -> CoolipyAPIResponse[MessageResponse]:
410        """Delete an environment variable by UUID."""
411        return await self._delete(
412            f"/databases/{uuid}/envs/{env_uuid}", response_model=MessageResponse
413        )
414
415    async def storages(self, uuid: str) -> CoolipyAPIResponse[dict]:
416        """List storages for a database."""
417        return await self._get(f"/databases/{uuid}/storages", response_model=dict)
418
419    async def create_storage(self, uuid: str, model: StorageCreate) -> CoolipyAPIResponse[dict]:
420        """Create a storage for a database."""
421        return await self._post(
422            f"/databases/{uuid}/storages", json=_dump(model), response_model=dict
423        )
424
425    async def update_storage(self, uuid: str, model: StorageUpdate) -> CoolipyAPIResponse[dict]:
426        """Update a storage for a database."""
427        return await self._patch(
428            f"/databases/{uuid}/storages", json=_dump(model), response_model=dict
429        )
430
431    async def delete_storage(
432        self, uuid: str, storage_uuid: str
433    ) -> CoolipyAPIResponse[MessageResponse]:
434        """Delete a storage by UUID."""
435        return await self._delete(
436            f"/databases/{uuid}/storages/{storage_uuid}", response_model=MessageResponse
437        )
438
439    async def tags(self, uuid: str) -> CoolipyAPIResponse[TagList]:
440        """List tags for a database."""
441        return await self._get(f"/databases/{uuid}/tags", response_model=list[Tag])
442
443    async def add_tags(self, uuid: str, model: TagsCreate) -> CoolipyAPIResponse[TagList]:
444        """Add one or more tags to a database."""
445        return await self._post(
446            f"/databases/{uuid}/tags", json=_dump(model), response_model=list[Tag]
447        )
448
449    async def delete_tag(self, uuid: str, tag_uuid: str) -> CoolipyAPIResponse[Any]:
450        """Remove a tag from a database."""
451        return await self._delete(f"/databases/{uuid}/tags/{tag_uuid}")
452
453    async def clone(
454        self,
455        uuid: str,
456        destination_uuid: str,
457        *,
458        name: str | None = None,
459        clone_volumes: bool = False,
460    ) -> CoolipyAPIResponse[dict]:
461        """Clone a database into a destination."""
462        body = {"destination_uuid": destination_uuid, "name": name, "clone_volumes": clone_volumes}
463        return await self._post(f"/databases/{uuid}/clone", json=body, response_model=dict)
464
465    async def update_storage_backup(
466        self, uuid: str, storage_uuid: str, model: VolumeBackupScheduleRequest
467    ) -> CoolipyAPIResponse[VolumeBackupScheduleResponse]:
468        """Schedule backups for a storage volume."""
469        return await self._put(
470            f"/databases/{uuid}/storages/{storage_uuid}/backups",
471            json=_dump(model),
472            response_model=VolumeBackupScheduleResponse,
473        )
474
475    async def delete_storage_backup(
476        self, uuid: str, storage_uuid: str
477    ) -> CoolipyAPIResponse[MessageResponse]:
478        """Remove the backup schedule for a storage volume."""
479        return await self._delete(
480            f"/databases/{uuid}/storages/{storage_uuid}/backups", response_model=MessageResponse
481        )
482
483    async def run_storage_backup(
484        self, uuid: str, storage_uuid: str
485    ) -> CoolipyAPIResponse[MessageResponse]:
486        """Run a storage backup now."""
487        return await self._post(
488            f"/databases/{uuid}/storages/{storage_uuid}/backups/run", response_model=MessageResponse
489        )