(self, request)
| 1422 | responses={200: inline_serializer(name="PluginRepoSettingsUpdated", fields={"refresh_interval_hours": serializers.IntegerField()})}, |
| 1423 | ) |
| 1424 | def put(self, request): |
| 1425 | from core.models import CoreSettings |
| 1426 | from core.scheduling import create_or_update_periodic_task, delete_periodic_task |
| 1427 | from .tasks import PLUGIN_REPO_REFRESH_TASK_NAME |
| 1428 | |
| 1429 | interval = request.data.get("refresh_interval_hours", 6) |
| 1430 | try: |
| 1431 | interval = int(interval) |
| 1432 | if interval < 0: |
| 1433 | interval = 0 |
| 1434 | except (TypeError, ValueError): |
| 1435 | interval = 6 |
| 1436 | |
| 1437 | obj, _ = CoreSettings.objects.update_or_create( |
| 1438 | key="plugin_repo_settings", |
| 1439 | defaults={ |
| 1440 | "name": "Plugin Repo Settings", |
| 1441 | "value": {"refresh_interval_hours": interval}, |
| 1442 | }, |
| 1443 | ) |
| 1444 | |
| 1445 | if interval == 0: |
| 1446 | delete_periodic_task(PLUGIN_REPO_REFRESH_TASK_NAME) |
| 1447 | else: |
| 1448 | create_or_update_periodic_task( |
| 1449 | task_name=PLUGIN_REPO_REFRESH_TASK_NAME, |
| 1450 | celery_task_path="apps.plugins.tasks.refresh_plugin_repos", |
| 1451 | interval_hours=interval, |
| 1452 | enabled=True, |
| 1453 | ) |
| 1454 | |
| 1455 | return Response(obj.value) |
nothing calls this directly
no test coverage detected