coolipy.resources.services

  1"""Services 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    BulkEnvsUpdate,
 11    DeploymentQueuedResponse,
 12    EnvironmentVariable,
 13    EnvironmentVariableCreate,
 14    EnvironmentVariableUpdate,
 15    Logs,
 16    MessageResponse,
 17    ScheduledTask,
 18    ScheduledTaskCreate,
 19    ScheduledTaskExecution,
 20    ScheduledTaskUpdate,
 21    StorageCreate,
 22    StorageUpdate,
 23    Tag,
 24    TagsCreate,
 25    UUIDResponse,
 26    VolumeBackupScheduleRequest,
 27    VolumeBackupScheduleResponse,
 28)
 29from coolipy.models.services import ServiceCreateModel, ServiceModel, ServiceUpdateModel
 30
 31ServiceList = list[ServiceModel]
 32EnvironmentVariableList = list[EnvironmentVariable]
 33ScheduledTaskList = list[ScheduledTask]
 34ScheduledTaskExecutionList = list[ScheduledTaskExecution]
 35TagList = list[Tag]
 36
 37
 38def _dump(model: Any) -> dict[str, Any]:
 39    return model.model_dump(mode="json", exclude_none=True)
 40
 41
 42class Services(ResourceBase):
 43    """Synchronous client for Coolify services."""
 44
 45    def list(self) -> CoolipyAPIResponse[ServiceList]:
 46        """List all services."""
 47        return self._get("/services", response_model=list[ServiceModel])
 48
 49    def get(self, uuid: str) -> CoolipyAPIResponse[ServiceModel]:
 50        """Get a service by UUID."""
 51        return self._get(f"/services/{uuid}", response_model=ServiceModel)
 52
 53    def create(self, model: ServiceCreateModel) -> CoolipyAPIResponse[dict]:
 54        """Create a service."""
 55        return self._post("/services", json=_dump(model), response_model=dict)
 56
 57    def update(self, uuid: str, model: ServiceUpdateModel) -> CoolipyAPIResponse[dict]:
 58        """Update a service by UUID."""
 59        return self._patch(f"/services/{uuid}", json=_dump(model), response_model=dict)
 60
 61    def delete(
 62        self,
 63        uuid: str,
 64        *,
 65        delete_configurations: bool = True,
 66        delete_volumes: bool = True,
 67        docker_cleanup: bool = True,
 68        delete_connected_networks: bool = True,
 69    ) -> CoolipyAPIResponse[MessageResponse]:
 70        """Delete a service by UUID."""
 71        params = {
 72            "delete_configurations": delete_configurations,
 73            "delete_volumes": delete_volumes,
 74            "docker_cleanup": docker_cleanup,
 75            "delete_connected_networks": delete_connected_networks,
 76        }
 77        return self._delete(f"/services/{uuid}", params=params, response_model=MessageResponse)
 78
 79    def logs(self, uuid: str) -> CoolipyAPIResponse[Logs]:
 80        """Get service logs."""
 81        return self._get(f"/services/{uuid}/logs", response_model=Logs)
 82
 83    def envs(self, uuid: str) -> CoolipyAPIResponse[EnvironmentVariableList]:
 84        """List environment variables for a service."""
 85        return self._get(f"/services/{uuid}/envs", response_model=list[EnvironmentVariable])
 86
 87    def create_env(
 88        self, uuid: str, model: EnvironmentVariableCreate
 89    ) -> CoolipyAPIResponse[UUIDResponse]:
 90        """Create an environment variable for a service."""
 91        return self._post(f"/services/{uuid}/envs", json=_dump(model), response_model=UUIDResponse)
 92
 93    def update_env(
 94        self, uuid: str, model: EnvironmentVariableUpdate
 95    ) -> CoolipyAPIResponse[EnvironmentVariable]:
 96        """Update an environment variable for a service."""
 97        return self._patch(
 98            f"/services/{uuid}/envs", json=_dump(model), response_model=EnvironmentVariable
 99        )
100
101    def bulk_update_envs(
102        self, uuid: str, model: BulkEnvsUpdate
103    ) -> CoolipyAPIResponse[EnvironmentVariableList]:
104        """Bulk-update environment variables for a service."""
105        return self._patch(
106            f"/services/{uuid}/envs/bulk",
107            json=_dump(model),
108            response_model=list[EnvironmentVariable],
109        )
110
111    def delete_env(self, uuid: str, env_uuid: str) -> CoolipyAPIResponse[MessageResponse]:
112        """Delete an environment variable by UUID."""
113        return self._delete(f"/services/{uuid}/envs/{env_uuid}", response_model=MessageResponse)
114
115    def move(self, uuid: str, environment_uuid: str) -> CoolipyAPIResponse[dict]:
116        """Move a service to another environment."""
117        return self._post(
118            f"/services/{uuid}/move",
119            json={"environment_uuid": environment_uuid},
120            response_model=dict,
121        )
122
123    def migrate(
124        self, uuid: str, destination_uuid: str, *, migrate_volumes: bool = True
125    ) -> CoolipyAPIResponse[Any]:
126        """Migrate a service to another destination/server."""
127        body = {"destination_uuid": destination_uuid, "migrate_volumes": migrate_volumes}
128        return self._post(f"/services/{uuid}/migrate", json=body)
129
130    def start(self, uuid: str) -> CoolipyAPIResponse[DeploymentQueuedResponse]:
131        """Start a service."""
132        return self._post(f"/services/{uuid}/start", response_model=DeploymentQueuedResponse)
133
134    def stop(self, uuid: str) -> CoolipyAPIResponse[MessageResponse]:
135        """Stop a service."""
136        return self._post(f"/services/{uuid}/stop", response_model=MessageResponse)
137
138    def restart(self, uuid: str) -> CoolipyAPIResponse[DeploymentQueuedResponse]:
139        """Restart a service."""
140        return self._post(f"/services/{uuid}/restart", response_model=DeploymentQueuedResponse)
141
142    def storages(self, uuid: str) -> CoolipyAPIResponse[dict]:
143        """List storages for a service."""
144        return self._get(f"/services/{uuid}/storages", response_model=dict)
145
146    def create_storage(self, uuid: str, model: StorageCreate) -> CoolipyAPIResponse[dict]:
147        """Create a storage for a service."""
148        return self._post(f"/services/{uuid}/storages", json=_dump(model), response_model=dict)
149
150    def update_storage(self, uuid: str, model: StorageUpdate) -> CoolipyAPIResponse[dict]:
151        """Update a storage for a service."""
152        return self._patch(f"/services/{uuid}/storages", json=_dump(model), response_model=dict)
153
154    def delete_storage(self, uuid: str, storage_uuid: str) -> CoolipyAPIResponse[MessageResponse]:
155        """Delete a storage by UUID."""
156        return self._delete(
157            f"/services/{uuid}/storages/{storage_uuid}", response_model=MessageResponse
158        )
159
160    def tags(self, uuid: str) -> CoolipyAPIResponse[TagList]:
161        """List tags for a service."""
162        return self._get(f"/services/{uuid}/tags", response_model=list[Tag])
163
164    def add_tags(self, uuid: str, model: TagsCreate) -> CoolipyAPIResponse[TagList]:
165        """Add one or more tags to a service."""
166        return self._post(f"/services/{uuid}/tags", json=_dump(model), response_model=list[Tag])
167
168    def delete_tag(self, uuid: str, tag_uuid: str) -> CoolipyAPIResponse[Any]:
169        """Remove a tag from a service."""
170        return self._delete(f"/services/{uuid}/tags/{tag_uuid}")
171
172    def clone(
173        self,
174        uuid: str,
175        destination_uuid: str,
176        *,
177        name: str | None = None,
178        clone_volumes: bool = False,
179    ) -> CoolipyAPIResponse[dict]:
180        """Clone a service into a destination."""
181        body = {"destination_uuid": destination_uuid, "name": name, "clone_volumes": clone_volumes}
182        return self._post(f"/services/{uuid}/clone", json=body, response_model=dict)
183
184    def scheduled_tasks(self, uuid: str) -> CoolipyAPIResponse[ScheduledTaskList]:
185        """List scheduled tasks for a service."""
186        return self._get(f"/services/{uuid}/scheduled-tasks", response_model=list[ScheduledTask])
187
188    def create_scheduled_task(
189        self, uuid: str, model: ScheduledTaskCreate
190    ) -> CoolipyAPIResponse[ScheduledTask]:
191        """Create a scheduled task for a service."""
192        return self._post(
193            f"/services/{uuid}/scheduled-tasks", json=_dump(model), response_model=ScheduledTask
194        )
195
196    def update_scheduled_task(
197        self, uuid: str, task_uuid: str, model: ScheduledTaskUpdate
198    ) -> CoolipyAPIResponse[ScheduledTask]:
199        """Update a scheduled task by UUID."""
200        return self._patch(
201            f"/services/{uuid}/scheduled-tasks/{task_uuid}",
202            json=_dump(model),
203            response_model=ScheduledTask,
204        )
205
206    def delete_scheduled_task(
207        self, uuid: str, task_uuid: str
208    ) -> CoolipyAPIResponse[MessageResponse]:
209        """Delete a scheduled task by UUID."""
210        return self._delete(
211            f"/services/{uuid}/scheduled-tasks/{task_uuid}", response_model=MessageResponse
212        )
213
214    def scheduled_task_executions(
215        self, uuid: str, task_uuid: str
216    ) -> CoolipyAPIResponse[ScheduledTaskExecutionList]:
217        """List executions of a scheduled task."""
218        return self._get(
219            f"/services/{uuid}/scheduled-tasks/{task_uuid}/executions",
220            response_model=list[ScheduledTaskExecution],
221        )
222
223    def execute_scheduled_task(
224        self, uuid: str, task_uuid: str
225    ) -> CoolipyAPIResponse[MessageResponse]:
226        """Execute a scheduled task now."""
227        return self._post(
228            f"/services/{uuid}/scheduled-tasks/{task_uuid}/execute", response_model=MessageResponse
229        )
230
231    def applications(self, uuid: str) -> CoolipyAPIResponse[Any]:
232        """List applications belonging to a service."""
233        return self._get(f"/services/{uuid}/applications")
234
235    def get_application(self, uuid: str, app_uuid: str) -> CoolipyAPIResponse[Any]:
236        """Get a service application by UUID."""
237        return self._get(f"/services/{uuid}/applications/{app_uuid}")
238
239    def update_application(self, uuid: str, app_uuid: str, model: Any) -> CoolipyAPIResponse[Any]:
240        """Update a service application."""
241        return self._patch(f"/services/{uuid}/applications/{app_uuid}", json=_dump(model))
242
243    def application_logs(self, uuid: str, app_uuid: str) -> CoolipyAPIResponse[Any]:
244        """Get service application logs."""
245        return self._get(f"/services/{uuid}/applications/{app_uuid}/logs")
246
247    def start_application(self, uuid: str, app_uuid: str) -> CoolipyAPIResponse[Any]:
248        """Start a service application."""
249        return self._post(f"/services/{uuid}/applications/{app_uuid}/start")
250
251    def restart_application(self, uuid: str, app_uuid: str) -> CoolipyAPIResponse[Any]:
252        """Restart a service application."""
253        return self._post(f"/services/{uuid}/applications/{app_uuid}/restart")
254
255    def stop_application(self, uuid: str, app_uuid: str) -> CoolipyAPIResponse[Any]:
256        """Stop a service application."""
257        return self._post(f"/services/{uuid}/applications/{app_uuid}/stop")
258
259    def databases(self, uuid: str) -> CoolipyAPIResponse[Any]:
260        """List databases belonging to a service."""
261        return self._get(f"/services/{uuid}/databases")
262
263    def get_database(self, uuid: str, database_uuid: str) -> CoolipyAPIResponse[Any]:
264        """Get a service database by UUID."""
265        return self._get(f"/services/{uuid}/databases/{database_uuid}")
266
267    def update_database(self, uuid: str, database_uuid: str, model: Any) -> CoolipyAPIResponse[Any]:
268        """Update a service database."""
269        return self._patch(f"/services/{uuid}/databases/{database_uuid}", json=_dump(model))
270
271    def database_logs(self, uuid: str, database_uuid: str) -> CoolipyAPIResponse[Any]:
272        """Get service database logs."""
273        return self._get(f"/services/{uuid}/databases/{database_uuid}/logs")
274
275    def start_database(self, uuid: str, database_uuid: str) -> CoolipyAPIResponse[Any]:
276        """Start a service database."""
277        return self._post(f"/services/{uuid}/databases/{database_uuid}/start")
278
279    def restart_database(self, uuid: str, database_uuid: str) -> CoolipyAPIResponse[Any]:
280        """Restart a service database."""
281        return self._post(f"/services/{uuid}/databases/{database_uuid}/restart")
282
283    def stop_database(self, uuid: str, database_uuid: str) -> CoolipyAPIResponse[Any]:
284        """Stop a service database."""
285        return self._post(f"/services/{uuid}/databases/{database_uuid}/stop")
286
287    def update_storage_backup(
288        self, uuid: str, storage_uuid: str, model: VolumeBackupScheduleRequest
289    ) -> CoolipyAPIResponse[VolumeBackupScheduleResponse]:
290        """Schedule backups for a storage volume."""
291        return self._put(
292            f"/services/{uuid}/storages/{storage_uuid}/backups",
293            json=_dump(model),
294            response_model=VolumeBackupScheduleResponse,
295        )
296
297    def delete_storage_backup(
298        self, uuid: str, storage_uuid: str
299    ) -> CoolipyAPIResponse[MessageResponse]:
300        """Remove the backup schedule for a storage volume."""
301        return self._delete(
302            f"/services/{uuid}/storages/{storage_uuid}/backups", response_model=MessageResponse
303        )
304
305    def run_storage_backup(
306        self, uuid: str, storage_uuid: str
307    ) -> CoolipyAPIResponse[MessageResponse]:
308        """Run a storage backup now."""
309        return self._post(
310            f"/services/{uuid}/storages/{storage_uuid}/backups/run", response_model=MessageResponse
311        )
312
313
314class AsyncServices(AsyncResourceBase):
315    """Asynchronous client for Coolify services."""
316
317    async def list(self) -> CoolipyAPIResponse[ServiceList]:
318        """List all services."""
319        return await self._get("/services", response_model=list[ServiceModel])
320
321    async def get(self, uuid: str) -> CoolipyAPIResponse[ServiceModel]:
322        """Get a service by UUID."""
323        return await self._get(f"/services/{uuid}", response_model=ServiceModel)
324
325    async def create(self, model: ServiceCreateModel) -> CoolipyAPIResponse[dict]:
326        """Create a service."""
327        return await self._post("/services", json=_dump(model), response_model=dict)
328
329    async def update(self, uuid: str, model: ServiceUpdateModel) -> CoolipyAPIResponse[dict]:
330        """Update a service by UUID."""
331        return await self._patch(f"/services/{uuid}", json=_dump(model), response_model=dict)
332
333    async def delete(
334        self,
335        uuid: str,
336        *,
337        delete_configurations: bool = True,
338        delete_volumes: bool = True,
339        docker_cleanup: bool = True,
340        delete_connected_networks: bool = True,
341    ) -> CoolipyAPIResponse[MessageResponse]:
342        """Delete a service by UUID."""
343        params = {
344            "delete_configurations": delete_configurations,
345            "delete_volumes": delete_volumes,
346            "docker_cleanup": docker_cleanup,
347            "delete_connected_networks": delete_connected_networks,
348        }
349        return await self._delete(
350            f"/services/{uuid}", params=params, response_model=MessageResponse
351        )
352
353    async def logs(self, uuid: str) -> CoolipyAPIResponse[Logs]:
354        """Get service logs."""
355        return await self._get(f"/services/{uuid}/logs", response_model=Logs)
356
357    async def envs(self, uuid: str) -> CoolipyAPIResponse[EnvironmentVariableList]:
358        """List environment variables for a service."""
359        return await self._get(f"/services/{uuid}/envs", response_model=list[EnvironmentVariable])
360
361    async def create_env(
362        self, uuid: str, model: EnvironmentVariableCreate
363    ) -> CoolipyAPIResponse[UUIDResponse]:
364        """Create an environment variable for a service."""
365        return await self._post(
366            f"/services/{uuid}/envs", json=_dump(model), response_model=UUIDResponse
367        )
368
369    async def update_env(
370        self, uuid: str, model: EnvironmentVariableUpdate
371    ) -> CoolipyAPIResponse[EnvironmentVariable]:
372        """Update an environment variable for a service."""
373        return await self._patch(
374            f"/services/{uuid}/envs", json=_dump(model), response_model=EnvironmentVariable
375        )
376
377    async def bulk_update_envs(
378        self, uuid: str, model: BulkEnvsUpdate
379    ) -> CoolipyAPIResponse[EnvironmentVariableList]:
380        """Bulk-update environment variables for a service."""
381        return await self._patch(
382            f"/services/{uuid}/envs/bulk",
383            json=_dump(model),
384            response_model=list[EnvironmentVariable],
385        )
386
387    async def delete_env(self, uuid: str, env_uuid: str) -> CoolipyAPIResponse[MessageResponse]:
388        """Delete an environment variable by UUID."""
389        return await self._delete(
390            f"/services/{uuid}/envs/{env_uuid}", response_model=MessageResponse
391        )
392
393    async def move(self, uuid: str, environment_uuid: str) -> CoolipyAPIResponse[dict]:
394        """Move a service to another environment."""
395        return await self._post(
396            f"/services/{uuid}/move",
397            json={"environment_uuid": environment_uuid},
398            response_model=dict,
399        )
400
401    async def migrate(
402        self, uuid: str, destination_uuid: str, *, migrate_volumes: bool = True
403    ) -> CoolipyAPIResponse[Any]:
404        """Migrate a service to another destination/server."""
405        body = {"destination_uuid": destination_uuid, "migrate_volumes": migrate_volumes}
406        return await self._post(f"/services/{uuid}/migrate", json=body)
407
408    async def start(self, uuid: str) -> CoolipyAPIResponse[DeploymentQueuedResponse]:
409        """Start a service."""
410        return await self._post(f"/services/{uuid}/start", response_model=DeploymentQueuedResponse)
411
412    async def stop(self, uuid: str) -> CoolipyAPIResponse[MessageResponse]:
413        """Stop a service."""
414        return await self._post(f"/services/{uuid}/stop", response_model=MessageResponse)
415
416    async def restart(self, uuid: str) -> CoolipyAPIResponse[DeploymentQueuedResponse]:
417        """Restart a service."""
418        return await self._post(
419            f"/services/{uuid}/restart", response_model=DeploymentQueuedResponse
420        )
421
422    async def storages(self, uuid: str) -> CoolipyAPIResponse[dict]:
423        """List storages for a service."""
424        return await self._get(f"/services/{uuid}/storages", response_model=dict)
425
426    async def create_storage(self, uuid: str, model: StorageCreate) -> CoolipyAPIResponse[dict]:
427        """Create a storage for a service."""
428        return await self._post(
429            f"/services/{uuid}/storages", json=_dump(model), response_model=dict
430        )
431
432    async def update_storage(self, uuid: str, model: StorageUpdate) -> CoolipyAPIResponse[dict]:
433        """Update a storage for a service."""
434        return await self._patch(
435            f"/services/{uuid}/storages", json=_dump(model), response_model=dict
436        )
437
438    async def delete_storage(
439        self, uuid: str, storage_uuid: str
440    ) -> CoolipyAPIResponse[MessageResponse]:
441        """Delete a storage by UUID."""
442        return await self._delete(
443            f"/services/{uuid}/storages/{storage_uuid}", response_model=MessageResponse
444        )
445
446    async def tags(self, uuid: str) -> CoolipyAPIResponse[TagList]:
447        """List tags for a service."""
448        return await self._get(f"/services/{uuid}/tags", response_model=list[Tag])
449
450    async def add_tags(self, uuid: str, model: TagsCreate) -> CoolipyAPIResponse[TagList]:
451        """Add one or more tags to a service."""
452        return await self._post(
453            f"/services/{uuid}/tags", json=_dump(model), response_model=list[Tag]
454        )
455
456    async def delete_tag(self, uuid: str, tag_uuid: str) -> CoolipyAPIResponse[Any]:
457        """Remove a tag from a service."""
458        return await self._delete(f"/services/{uuid}/tags/{tag_uuid}")
459
460    async def clone(
461        self,
462        uuid: str,
463        destination_uuid: str,
464        *,
465        name: str | None = None,
466        clone_volumes: bool = False,
467    ) -> CoolipyAPIResponse[dict]:
468        """Clone a service into a destination."""
469        body = {"destination_uuid": destination_uuid, "name": name, "clone_volumes": clone_volumes}
470        return await self._post(f"/services/{uuid}/clone", json=body, response_model=dict)
471
472    async def scheduled_tasks(self, uuid: str) -> CoolipyAPIResponse[ScheduledTaskList]:
473        """List scheduled tasks for a service."""
474        return await self._get(
475            f"/services/{uuid}/scheduled-tasks", response_model=list[ScheduledTask]
476        )
477
478    async def create_scheduled_task(
479        self, uuid: str, model: ScheduledTaskCreate
480    ) -> CoolipyAPIResponse[ScheduledTask]:
481        """Create a scheduled task for a service."""
482        return await self._post(
483            f"/services/{uuid}/scheduled-tasks", json=_dump(model), response_model=ScheduledTask
484        )
485
486    async def update_scheduled_task(
487        self, uuid: str, task_uuid: str, model: ScheduledTaskUpdate
488    ) -> CoolipyAPIResponse[ScheduledTask]:
489        """Update a scheduled task by UUID."""
490        return await self._patch(
491            f"/services/{uuid}/scheduled-tasks/{task_uuid}",
492            json=_dump(model),
493            response_model=ScheduledTask,
494        )
495
496    async def delete_scheduled_task(
497        self, uuid: str, task_uuid: str
498    ) -> CoolipyAPIResponse[MessageResponse]:
499        """Delete a scheduled task by UUID."""
500        return await self._delete(
501            f"/services/{uuid}/scheduled-tasks/{task_uuid}", response_model=MessageResponse
502        )
503
504    async def scheduled_task_executions(
505        self, uuid: str, task_uuid: str
506    ) -> CoolipyAPIResponse[ScheduledTaskExecutionList]:
507        """List executions of a scheduled task."""
508        return await self._get(
509            f"/services/{uuid}/scheduled-tasks/{task_uuid}/executions",
510            response_model=list[ScheduledTaskExecution],
511        )
512
513    async def execute_scheduled_task(
514        self, uuid: str, task_uuid: str
515    ) -> CoolipyAPIResponse[MessageResponse]:
516        """Execute a scheduled task now."""
517        return await self._post(
518            f"/services/{uuid}/scheduled-tasks/{task_uuid}/execute", response_model=MessageResponse
519        )
520
521    async def applications(self, uuid: str) -> CoolipyAPIResponse[Any]:
522        """List applications belonging to a service."""
523        return await self._get(f"/services/{uuid}/applications")
524
525    async def get_application(self, uuid: str, app_uuid: str) -> CoolipyAPIResponse[Any]:
526        """Get a service application by UUID."""
527        return await self._get(f"/services/{uuid}/applications/{app_uuid}")
528
529    async def update_application(
530        self, uuid: str, app_uuid: str, model: Any
531    ) -> CoolipyAPIResponse[Any]:
532        """Update a service application."""
533        return await self._patch(f"/services/{uuid}/applications/{app_uuid}", json=_dump(model))
534
535    async def application_logs(self, uuid: str, app_uuid: str) -> CoolipyAPIResponse[Any]:
536        """Get service application logs."""
537        return await self._get(f"/services/{uuid}/applications/{app_uuid}/logs")
538
539    async def start_application(self, uuid: str, app_uuid: str) -> CoolipyAPIResponse[Any]:
540        """Start a service application."""
541        return await self._post(f"/services/{uuid}/applications/{app_uuid}/start")
542
543    async def restart_application(self, uuid: str, app_uuid: str) -> CoolipyAPIResponse[Any]:
544        """Restart a service application."""
545        return await self._post(f"/services/{uuid}/applications/{app_uuid}/restart")
546
547    async def stop_application(self, uuid: str, app_uuid: str) -> CoolipyAPIResponse[Any]:
548        """Stop a service application."""
549        return await self._post(f"/services/{uuid}/applications/{app_uuid}/stop")
550
551    async def databases(self, uuid: str) -> CoolipyAPIResponse[Any]:
552        """List databases belonging to a service."""
553        return await self._get(f"/services/{uuid}/databases")
554
555    async def get_database(self, uuid: str, database_uuid: str) -> CoolipyAPIResponse[Any]:
556        """Get a service database by UUID."""
557        return await self._get(f"/services/{uuid}/databases/{database_uuid}")
558
559    async def update_database(
560        self, uuid: str, database_uuid: str, model: Any
561    ) -> CoolipyAPIResponse[Any]:
562        """Update a service database."""
563        return await self._patch(f"/services/{uuid}/databases/{database_uuid}", json=_dump(model))
564
565    async def database_logs(self, uuid: str, database_uuid: str) -> CoolipyAPIResponse[Any]:
566        """Get service database logs."""
567        return await self._get(f"/services/{uuid}/databases/{database_uuid}/logs")
568
569    async def start_database(self, uuid: str, database_uuid: str) -> CoolipyAPIResponse[Any]:
570        """Start a service database."""
571        return await self._post(f"/services/{uuid}/databases/{database_uuid}/start")
572
573    async def restart_database(self, uuid: str, database_uuid: str) -> CoolipyAPIResponse[Any]:
574        """Restart a service database."""
575        return await self._post(f"/services/{uuid}/databases/{database_uuid}/restart")
576
577    async def stop_database(self, uuid: str, database_uuid: str) -> CoolipyAPIResponse[Any]:
578        """Stop a service database."""
579        return await self._post(f"/services/{uuid}/databases/{database_uuid}/stop")
580
581    async def update_storage_backup(
582        self, uuid: str, storage_uuid: str, model: VolumeBackupScheduleRequest
583    ) -> CoolipyAPIResponse[VolumeBackupScheduleResponse]:
584        """Schedule backups for a storage volume."""
585        return await self._put(
586            f"/services/{uuid}/storages/{storage_uuid}/backups",
587            json=_dump(model),
588            response_model=VolumeBackupScheduleResponse,
589        )
590
591    async def delete_storage_backup(
592        self, uuid: str, storage_uuid: str
593    ) -> CoolipyAPIResponse[MessageResponse]:
594        """Remove the backup schedule for a storage volume."""
595        return await self._delete(
596            f"/services/{uuid}/storages/{storage_uuid}/backups", response_model=MessageResponse
597        )
598
599    async def run_storage_backup(
600        self, uuid: str, storage_uuid: str
601    ) -> CoolipyAPIResponse[MessageResponse]:
602        """Run a storage backup now."""
603        return await self._post(
604            f"/services/{uuid}/storages/{storage_uuid}/backups/run", response_model=MessageResponse
605        )