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.dao import TaskStatus 

12from seedboxsync.front.apis import DateTimeOrZero, Resource 

13 

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

15 

16 

17# ========================== 

18# Models 

19# ========================== 

20taskstatus_model = api.model( 

21 "TaskStatuses", 

22 { 

23 "key": fields.String( 

24 required=True, 

25 description="Unique task identifier", 

26 example="sync_blackhole", 

27 ), 

28 "running": fields.Boolean( 

29 dt_format="iso8601", 

30 required=True, 

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

32 example=False, 

33 ), 

34 "started": DateTimeOrZero( 

35 dt_format="iso8601", 

36 required=False, 

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

38 ), 

39 "finished": DateTimeOrZero( 

40 dt_format="iso8601", 

41 required=True, 

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

43 ), 

44 }, 

45) 

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

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

48 

49 

50# ========================== 

51# Endpoints 

52# ========================== 

53@api.route("") 

54class TaskStatusList(Resource): 

55 """ 

56 Endpoint for managing taskstatus list. 

57 

58 Provides a list of taskstatus. 

59 """ 

60 

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

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

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

64 """Retrieve a list of taskstatus.""" 

65 select = TaskStatus.select( 

66 TaskStatus.key, 

67 TaskStatus.running, 

68 TaskStatus.started, 

69 TaskStatus.finished, 

70 ) 

71 

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

73 

74 

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

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

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

78class TaskStatuses(Resource): 

79 """ 

80 Endpoint for managing taskstatuses. 

81 

82 Provides taskstatuses operations. 

83 """ 

84 

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

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

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

88 """ 

89 Retrieve a taskstatus. 

90 

91 Args: 

92 key (str): TaskStatus identifier. 

93 

94 Returns: 

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

96 """ 

97 select: TaskStatus | None = None 

98 try: 

99 select = ( 

100 TaskStatus.select( 

101 TaskStatus.key, 

102 TaskStatus.running, 

103 TaskStatus.started, 

104 TaskStatus.finished, 

105 ) 

106 .where(TaskStatus.key == key) 

107 .dicts() 

108 .get() 

109 ) 

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

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

112 

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