Coverage for utils/response.py: 87.50%

1from datetime import datetime 

2from pydantic import BaseModel, RootModel 

3from typing import Optional, Type, TypeVar, Generic 

4 

5T = TypeVar("T") 

6 

7class APIResponse(BaseModel, Generic[T]): 

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

9 code: int 

10 message: str 

11 data: Optional[T] = None 

12 

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

14 """Model for validation error data structure""" 

15 pass 

16 

17def make_response_doc(description: str, model: Optional[Type] = None, example: Optional[dict] = None) -> dict: 

18 """Create OpenAPI response documentation with model and example""" 

19 doc = {"description": description} 

20 if model: 

21 doc["model"] = APIResponse[model] 

22 if example: 

23 doc["content"] = {"application/json": {"example": example}} 

24 return doc 

25 

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

27 """ 

28 Parse and merge responses. Supports: 

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

30 - 3-tuple: (description, model, example) - uses provided example  

31 - string: description only - creates simple error response 

32 """ 

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

34 merged = {} 

35 if default: 

36 merged.update(default) 

37 if custom: 

38 merged.update(custom) 

39 

40 result = {} 

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

42 if isinstance(val, tuple): 

43 if len(val) == 2: 

44 desc, model = val 

45 if model is None: 

46 data_example = None 

47 else: 

48 try: 

49 schema = model.model_json_schema() 

50 data_example = generate_example_from_schema(schema) 

51 except: 

52 data_example = None 

53 

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

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

56 elif len(val) == 3: 

57 desc, model, example = val 

58 # Auto-fill missing code and message 

59 if "code" not in example: 

60 example["code"] = code 

61 if "message" not in example: 

62 example["message"] = desc 

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

64 elif isinstance(val, str): 

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

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

67 else: 

68 result[code] = val 

69 return result 

70 

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

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

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

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

75 example = {} 

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

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

78 return example 

79 return None 

80 

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

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

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

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

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

86 if referenced_schema: 

87 return generate_example_from_schema(referenced_schema) 

88 return None 

89 

90 prop_type = prop.get("type") 

91 

92 if prop_type == "string": 

93 if key == "id": 

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

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

96 return "user@example.com" 

97 elif key == "phone": 

98 return "123456789" 

99 elif key == "created_at": 

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

101 elif key == "updated_at": 

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

103 elif key == "expires_at": 

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

105 else: 

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

107 elif prop_type == "integer": 

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

109 return 10 

110 elif key == "page": 

111 return 1 

112 else: 

113 return 100 

114 elif prop_type == "number": 

115 return 123.45 

116 elif prop_type == "boolean": 

117 return True 

118 elif prop_type == "array": 

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

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

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

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

123 if referenced_schema: 

124 item_example = generate_example_from_schema(referenced_schema) 

125 return [item_example] if item_example else [] 

126 return [] 

127 elif prop_type == "object": 

128 return generate_example_from_schema(prop) 

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

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

131 elif prop.get("anyOf"): 

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

133 for option in options: 

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

135 return generate_property_example(option, key, full_schema) 

136 return None 

137 else: 

138 return None 

139 

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

141 """ 

142 Resolve JSON Schema $ref references to actual schema definitions 

143 """ 

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

145 return None 

146 

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

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

149 current = schema 

150 

151 # Navigate through nested dict structure following the path 

152 for part in path_parts: 

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

154 current = current[part] 

155 else: 

156 # Reference not found 

157 return None 

158 

159 if isinstance(current, dict): 

160 return current 

161 else: 

162 return None 

163 

164common_responses = { 

165 401: ( 

166 "Invalid or expired token", 

167 APIResponse[None], 

168 { 

169 "code": 401, 

170 "message": "Invalid or expired token", 

171 "data": None 

172 } 

173 ), 

174 403: ( 

175 "Permission denied", 

176 APIResponse[None], 

177 { 

178 "code": 403, 

179 "message": "Permission denied", 

180 "data": None 

181 } 

182 ), 

183 422: ( 

184 "Validation Error", 

185 APIResponse[ValidationErrorData], 

186 { 

187 "code": 422, 

188 "message": "Validation Error", 

189 "data": {"body.params": "field required"} 

190 } 

191 ), 

192 429: ( 

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

194 APIResponse[None], 

195 { 

196 "code": 429, 

197 "message": "Too many requests. Try again later.", 

198 "data": None 

199 } 

200 ), 

201 500: ( 

202 "Internal Server Error", 

203 APIResponse[None], 

204 { 

205 "code": 500, 

206 "message": "Internal Server Error", 

207 "data": None 

208 } 

209 ) 

210}