Coverage for utils/response.py: 100.00%

1from datetime import datetime 

2from typing import TypeVar 

3 

4from pydantic import BaseModel, RootModel 

5 

6T = TypeVar("T") 

7 

8 

9class APIResponse[T](BaseModel): 

10 """Generic API response wrapper that maintains consistent response structure""" 

11 

12 code: int 

13 message: str 

14 data: T | None = None 

15 

16 

17class ValidationErrorData(RootModel[dict[str, str]]): 

18 """Model for validation error data structure""" 

19 

20 pass 

21 

22 

23def is_openapi_examples(example: dict) -> bool: 

24 """ 

25 Detect OpenAPI Media Type ``examples`` (named map with dropdown in Swagger UI). 

26 

27 Expected shape: 

28 { 

29 "caseA": {"summary": "...", "value": {...}}, 

30 "caseB": {"value": {...}}, 

31 } 

32 """ 

33 if not isinstance(example, dict) or not example: 

34 return False 

35 if "code" in example or "message" in example or "data" in example: 

36 return False 

37 return all(isinstance(item, dict) and "value" in item for item in example.values()) 

38 

39 

40def make_response_doc( 

41 description: str, model: type | None = None, example: dict | None = None 

42) -> dict: 

43 """Create OpenAPI response documentation with model and example(s)""" 

44 doc = {"description": description} 

45 if model: 

46 doc["model"] = APIResponse[model] 

47 if example: 

48 media = {"examples": example} if is_openapi_examples(example) else {"example": example} 

49 doc["content"] = {"application/json": media} 

50 return doc 

51 

52 

53def make_error_examples(code: int, cases: dict[str, str]) -> dict: 

54 """ 

55 Build OpenAPI named ``examples`` for simple error responses. 

56 

57 Example: 

58 make_error_examples(401, { 

59 "invalidSession": "Invalid or expired session", 

60 "invalidCsrf": "Invalid or expired CSRF token", 

61 }) 

62 """ 

63 return { 

64 key: { 

65 "summary": message, 

66 "value": {"code": code, "message": message, "data": None}, 

67 } 

68 for key, message in cases.items() 

69 } 

70 

71 

72def parse_responses(custom: dict, default: dict = None) -> dict: 

73 """ 

74 Parse and merge responses. Supports: 

75 - 2-tuple: (description, model) - auto-generates example 

76 - 3-tuple: (description, model, example) - single example dict, 

77 or OpenAPI named ``examples`` map (Swagger UI dropdown) 

78 - string: description only - creates simple error response 

79 """ 

80 # Merge default responses first, then override with custom ones 

81 merged = {} 

82 if default: 

83 merged.update(default) 

84 if custom: 

85 merged.update(custom) 

86 

87 result = {} 

88 for code, val in merged.items(): 

89 if isinstance(val, tuple): 

90 if len(val) == 2: 

91 desc, model = val 

92 if model is None: 

93 data_example = None 

94 else: 

95 try: 

96 schema = model.model_json_schema() 

97 data_example = generate_example_from_schema(schema) 

98 except Exception: 

99 data_example = None 

100 

101 example = {"code": code, "message": desc, "data": data_example} 

102 result[code] = make_response_doc(desc, model, example) 

103 elif len(val) == 3: 

104 desc, model, example = val 

105 # Auto-fill only for a single response body example 

106 if not is_openapi_examples(example): 

107 if "code" not in example: 

108 example["code"] = code 

109 if "message" not in example: 

110 example["message"] = desc 

111 result[code] = make_response_doc(desc, model, example) 

112 elif isinstance(val, str): 

113 example = {"code": code, "message": val, "data": None} 

114 result[code] = make_response_doc(val, None, example) 

115 else: 

116 result[code] = val 

117 return result 

118 

119 

120def generate_example_from_schema(schema: dict) -> dict: 

121 """Generate example data from JSON schema object properties""" 

122 if schema.get("type") == "object": 

123 properties = schema.get("properties", {}) 

124 example = {} 

125 for key, prop in properties.items(): 

126 example[key] = generate_property_example(prop, key, schema) 

127 return example 

128 return None 

129 

130 

131def generate_property_example(prop: dict, key: str = "", full_schema: dict = None): 

132 """Generate example value for a single property based on its type and field name""" 

133 # Handle $ref references first (for nested objects) 

134 if prop.get("$ref"): 

135 referenced_schema = resolve_ref(prop["$ref"], full_schema) 

136 if referenced_schema: 

137 return generate_example_from_schema(referenced_schema) 

138 return None 

139 

140 prop_type = prop.get("type") 

141 

142 if prop_type == "string": 

143 if key == "id": 

144 return "123e4567-e89b-12d3-a456-426614174000" 

145 elif "email" in key.lower(): 

146 return "user@example.com" 

147 elif key == "phone": 

148 return "123456789" 

149 elif key == "created_at": 

150 return datetime.now().astimezone().isoformat() 

151 elif key == "updated_at": 

152 return datetime.now().astimezone().isoformat() 

153 elif key == "expires_at": 

154 return datetime.now().astimezone().isoformat() 

155 else: 

156 return f"Example {key.replace('_', ' ').title()}" 

157 elif prop_type == "integer": 

158 if key in ["per_page", "total_pages"]: 

159 return 10 

160 elif key == "page": 

161 return 1 

162 else: 

163 return 100 

164 elif prop_type == "number": 

165 return 123.45 

166 elif prop_type == "boolean": 

167 return True 

168 elif prop_type == "array": 

169 items_schema = prop.get("items", {}) 

170 # Handle $ref references in array items (e.g., List[UserRead]) 

171 if items_schema.get("$ref"): 

172 referenced_schema = resolve_ref(items_schema["$ref"], full_schema) 

173 if referenced_schema: 

174 item_example = generate_example_from_schema(referenced_schema) 

175 return [item_example] if item_example else [] 

176 return [] 

177 elif prop_type == "object": 

178 return generate_example_from_schema(prop) 

179 elif prop.get("format") == "date-time": 

180 return datetime.now().astimezone().isoformat() 

181 elif prop.get("anyOf"): 

182 options = prop.get("anyOf", []) 

183 for option in options: 

184 if option.get("type") != "null": 

185 return generate_property_example(option, key, full_schema) 

186 return None 

187 else: 

188 return None 

189 

190 

191def resolve_ref(ref_path: str, schema: dict) -> dict: 

192 """ 

193 Resolve JSON Schema $ref references to actual schema definitions 

194 """ 

195 if not ref_path.startswith("#/"): 

196 return None 

197 

198 # Parse reference path: "#/$defs/UserRead" -> ["$defs", "UserRead"] 

199 path_parts = ref_path[2:].split("/") 

200 current = schema 

201 

202 # Navigate through nested dict structure following the path 

203 for part in path_parts: 

204 if isinstance(current, dict) and part in current: 

205 current = current[part] 

206 else: 

207 # Reference not found 

208 return None 

209 

210 if isinstance(current, dict): 

211 return current 

212 else: 

213 return None 

214 

215 

216common_responses = { 

217 401: ( 

218 "Invalid or expired token", 

219 APIResponse[None], 

220 {"code": 401, "message": "Invalid or expired token", "data": None}, 

221 ), 

222 403: ( 

223 "Permission denied", 

224 APIResponse[None], 

225 {"code": 403, "message": "Permission denied", "data": None}, 

226 ), 

227 422: ( 

228 "Validation Error", 

229 APIResponse[ValidationErrorData], 

230 {"code": 422, "message": "Validation Error", "data": {"body.params": "field required"}}, 

231 ), 

232 429: ( 

233 "Too many requests. Try again later.", 

234 APIResponse[None], 

235 {"code": 429, "message": "Too many requests. Try again later.", "data": None}, 

236 ), 

237 500: ( 

238 "Internal Server Error", 

239 APIResponse[None], 

240 {"code": 500, "message": "Internal Server Error", "data": None}, 

241 ), 

242}