Coverage for seedboxsync/front/apis/taskstatus.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 TaskStatus view.""" 

8 

9from typing import Any 

10from flask_restx import Namespace, fields 

11from seedboxsync.core.database.models import TaskStatus 

12from seedboxsync.front.apis import DateTimeOrZero, Resource 

13from seedboxsync.front.login_manager import login_required 

14 

15api = Namespace("taskstatuses", description="Operations related to taskstatus") 

16 

17 

18# ========================== 

19# Models 

20# ========================== 

21taskstatus_model = api.model( 

22 "TaskStatuses", 

23 { 

24 "key": fields.String( 

25 required=True, 

26 description="Unique task identifier", 

27 example="sync_blackhole", 

28 ), 

29 "running": fields.Boolean( 

30 dt_format="iso8601", 

31 required=True, 

32 description="Whether the task is currently running", 

33 example=False, 

34 ), 

35 "started": DateTimeOrZero( 

36 dt_format="iso8601", 

37 required=False, 

38 description="Timestamp when the task execution started", 

39 ), 

40 "finished": DateTimeOrZero( 

41 dt_format="iso8601", 

42 required=True, 

43 description="Timestamp when the task execution finished", 

44 ), 

45 }, 

46) 

47taskstatus_list_envelope = Resource.build_envelope_model(api, "TaskStatusList", nested_model=taskstatus_model) 

48taskstatus_envelope = Resource.build_envelope_model(api, "TaskStatus", nested_model=taskstatus_model, as_list=False) 

49 

50 

51# ========================== 

52# Endpoints 

53# ========================== 

54@api.route("") 

55class TaskStatusList(Resource): 

56 """ 

57 Endpoint for managing taskstatus list. 

58 

59 Provides a list of taskstatus. 

60 """ 

61 

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

63 @api.marshal_with(taskstatus_list_envelope, code=200, description="List of taskstatuses") # type: ignore[untyped-decorator] 

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

65 def get(self) -> dict[str, Any]: 

66 """Retrieve a list of taskstatus.""" 

67 select = TaskStatus.select( 

68 TaskStatus.key, 

69 TaskStatus.running, 

70 TaskStatus.started, 

71 TaskStatus.finished, 

72 ) 

73 

74 return self.build_envelope(list(select.dicts()), type="TaskStatus") 

75 

76 

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

78@api.response(404, "TaskStatus not found") 

79@api.param("key", "The taskstatus key") 

80class TaskStatuses(Resource): 

81 """ 

82 Endpoint for managing taskstatuses. 

83 

84 Provides taskstatuses operations. 

85 """ 

86 

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

88 @api.marshal_with(taskstatus_envelope, code=200, description="TaskStatus element") # type: ignore[untyped-decorator] 

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

90 def get(self, key: str) -> dict[str, Any]: 

91 """ 

92 Retrieve a taskstatus. 

93 

94 Args: 

95 key (str): TaskStatus identifier. 

96 

97 Returns: 

98 dict[str, Any]: API response envelope containing the taskstatus. 

99 """ 

100 select: TaskStatus | None = None 

101 try: 

102 select = ( 

103 TaskStatus.select( 

104 TaskStatus.key, 

105 TaskStatus.running, 

106 TaskStatus.started, 

107 TaskStatus.finished, 

108 ) 

109 .where(TaskStatus.key == key) 

110 .dicts() 

111 .get() 

112 ) 

113 except TaskStatus.DoesNotExist: # type: ignore[attr-defined] 

114 api.abort(404, f"TaskStatus {key} doesn't exist") 

115 

116 return self.build_envelope(select, type="TaskStatus")