Coverage for seedboxsync/front/apis/core/resources.py: 97%

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 front RestXResource module.""" 

8 

9from datetime import datetime 

10from typing import Any 

11import uuid 

12from flask_restx import Model, Namespace, OrderedModel, Resource as RestXResource, fields, inputs, reqparse 

13 

14 

15class Resource(RestXResource): # type: ignore[misc] 

16 """ 

17 Base Resource class extending Flask-RestX Resource. 

18 

19 Provides utility methods for: 

20 - enforcing limits on query parameters 

21 - building consistent API response envelopes 

22 - generating envelope models for Swagger documentation 

23 """ 

24 

25 def set_limit(self, limit: int) -> int: 

26 """ 

27 Clamp the limit value within the allowed range [5, 1000]. 

28 

29 Args: 

30 limit (int): The requested limit value. 

31 

32 Returns: 

33 int: The clamped limit within [5, 1000]. 

34 """ 

35 if limit < 5: 

36 return 5 

37 if limit > 1000: 

38 return 1000 

39 return limit 

40 

41 def build_envelope( 

42 self, 

43 data: Any, 

44 *, 

45 type: str = "about:blank", # noqa: A002 

46 status_code: int = 200, 

47 message: str | None = None, 

48 data_total: int | None = None, 

49 ) -> dict[str, Any]: 

50 """ 

51 Build a standard API response envelope. 

52 

53 Args: 

54 data (Any): The response payload. 

55 type (str): The resource type or endpoint identifier (default: 'about:blank'). 

56 status_code (int): HTTP status code (default: 200). 

57 message (str): Send message in place of data. 

58 data_total (int | None): Optional total number of items if the result is paginated. 

59 

60 Returns: 

61 dict[str, Any]: Structured API response containing metadata and payload. 

62 """ 

63 return { 

64 "type": type, 

65 "success": True, 

66 "status": status_code, 

67 "timestamp": datetime.now().astimezone().isoformat(), 

68 "traceId": str(uuid.uuid4()), 

69 **({"data": data} if data is not None else {}), 

70 **({"data_total": data_total} if data_total is not None else {}), 

71 **({"message": message} if message is not None else {}), 

72 "data": data, 

73 } 

74 

75 @staticmethod 

76 def build_envelope_model( 

77 api: Namespace, 

78 name: str, 

79 *, 

80 nested_model: Model | OrderedModel | None = None, 

81 as_list: bool = True, 

82 as_message: bool = False, 

83 ) -> Model | OrderedModel: 

84 """ 

85 Build a Flask-RestX model for a standard API response envelope. 

86 

87 This is used for Swagger documentation. You can generate an envelope 

88 for a single resource or for a list of resources. 

89 

90 Args: 

91 api (Namespace): The Flask-RestX namespace. 

92 name (str): The name of the nested resource model. 

93 nested_model (Model | OrderedModel): The Flask-RestX model representing the resource. 

94 as_list (bool): If True, the 'data' field will be a list of nested resources. 

95 as_message (bool): The response is a message only. 

96 

97 Returns: 

98 Model | OrderedModel: A new Flask-RestX model representing the envelope. 

99 """ 

100 if as_message is False: 

101 if as_list: 

102 nested_name = getattr(nested_model, "name", "item") 

103 data_field = fields.List( 

104 fields.Nested(nested_model), 

105 required=True, 

106 description=f"List of {nested_name} objects", 

107 ) 

108 else: 

109 data_field = fields.Nested(nested_model, required=True, description=f"The {name} object") 

110 data_total = fields.Integer(required=False, description=f"Total of {name} object", example=1) 

111 message_field = None 

112 else: 

113 data_field = None 

114 data_total = None 

115 message_field = fields.String(required=False, description="Response message", example="All is OK") 

116 

117 return api.model( 

118 f"Envelope[{name}]", 

119 { 

120 "type": fields.String( 

121 required=True, 

122 description="Resource type or endpoint identifier", 

123 example=name, 

124 ), 

125 "success": fields.Boolean( 

126 required=True, 

127 description="Indicates if the request succeeded", 

128 example=True, 

129 ), 

130 "status": fields.Integer(required=True, description="HTTP status code", example=200), 

131 "timestamp": fields.DateTime(required=True, description="Timestamp of the response"), 

132 "traceId": fields.String( 

133 required=True, 

134 description="Unique trace identifier", 

135 example="0a8ab95e-a463-424e-bc6d-505503bf200d", 

136 ), 

137 **({"data": data_field} if data_field is not None else {}), 

138 **({"data_total": data_total} if data_total is not None else {}), 

139 **({"message": message_field} if message_field is not None else {}), 

140 }, 

141 ) 

142 

143 

144class DateTimeOrZero(fields.DateTime): # type: ignore[misc] 

145 """Format date-time values while preserving the legacy zero value.""" 

146 

147 def format(self, value: int | datetime) -> Any: 

148 """ 

149 Format a date-time value or return zero unchanged. 

150 

151 Args: 

152 value (int | datetime): Date-time value or the legacy zero sentinel. 

153 

154 Returns: 

155 Any: Formatted date-time value or zero. 

156 """ 

157 if value == 0: 

158 return 0 

159 return super().format(value) 

160 

161 

162# ========================== 

163# Global parsers 

164# ========================== 

165parser_period = reqparse.RequestParser() 

166parser_period.add_argument( 

167 "start_date", 

168 type=inputs.date_from_iso8601, 

169 location="args", 

170 help="Start date for filtering in ISO 8601 format (e.g. YYYY-MM-DD)", 

171) 

172parser_period.add_argument( 

173 "end_date", 

174 type=inputs.date_from_iso8601, 

175 location="args", 

176 help="End date for filtering in ISO 8601 format (e.g. YYYY-MM-DD)", 

177)