Coverage for seedboxsync/front/apis/tasks.py: 100%

1# 

2# Copyright (C) 2015-2026 Guillaume Kulakowski <guillaume@kulakowski.fr> 

3# 

4# For the full copyright and license information, please view the LICENSE 

5# file that was distributed with this source code. 

6# 

7"""SeedboxSync API Task view.""" 

8 

9from typing import Any 

10from flask_restx import Namespace 

11from seedboxsync.front.apis import Resource 

12 

13api = Namespace("tasks", description="Operations related to task lauching") 

14 

15 

16TASK_HANDLERS = { 

17 "seedbox": "seedboxsync.core.taskmanager.task.task_sync_seedbox.sync_seedbox", 

18 "blackhole": "seedboxsync.core.taskmanager.task.task_sync_blackhole.sync_blackhole", 

19} 

20 

21 

22@api.route("/<string:key>") 

23@api.response(404, "Task not found") 

24@api.param("key", "The task key") 

25class Tasks(Resource): 

26 """Endpoint for managing task lauching.""" 

27 

28 @api.doc("post_task") # type: ignore[untyped-decorator] 

29 @api.response(202, "Task launched") # type: ignore[untyped-decorator] 

30 def post(self, key: str) -> tuple[dict[str, Any], int]: 

31 """ 

32 Launch a task associated with a task key. 

33 

34 Args: 

35 key (str): Task identifier. 

36 

37 Returns: 

38 tuple[dict[str, Any], int]: API response body and HTTP status code. 

39 """ 

40 handler_path = TASK_HANDLERS.get(key) 

41 if handler_path is None: 

42 api.abort(404, f"Task '{key}' not found") 

43 

44 assert handler_path is not None 

45 module_name, func_name = handler_path.rsplit(".", 1) 

46 module = __import__(module_name, fromlist=[func_name]) 

47 task_func = getattr(module, func_name) 

48 

49 # If these are Huey tasks, prefer enqueueing instead of calling directly. 

50 task_func() 

51 

52 return {"status": "queued", "task": key}, 202