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 

12from seedboxsync.front.login_manager import login_required 

13 

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

15 

16 

17TASK_HANDLERS = { 

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

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

20} 

21 

22 

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

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

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

26class Tasks(Resource): 

27 """Endpoint for managing task lauching.""" 

28 

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

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

31 @login_required # type: ignore[untyped-decorator] 

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

33 """ 

34 Launch a task associated with a task key. 

35 

36 Args: 

37 key (str): Task identifier. 

38 

39 Returns: 

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

41 """ 

42 handler_path = TASK_HANDLERS.get(key) 

43 if handler_path is None: 

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

45 

46 assert handler_path is not None 

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

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

49 task_func = getattr(module, func_name) 

50 

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

52 task_func() 

53 

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