coolipy.resources.applications

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