Coverage for webapp / endpoints / models.py: 94%

1# Packages 

2import flask 

3from flask import make_response 

4from canonicalwebteam.exceptions import ( 

5 StoreApiResponseErrorList, 

6 StoreApiResourceNotFound, 

7) 

8from canonicalwebteam.store_api.publishergw import PublisherGW 

9 

10# Local 

11from webapp.decorators import login_required, exchange_required 

12from webapp.helpers import api_publisher_session, get_brand_id 

13 

14 

15publisher_gateway = PublisherGW("snap", api_publisher_session) 

16 

17models = flask.Blueprint( 

18 "models", 

19 __name__, 

20) 

21 

22 

23@models.route("/api/store/<store_id>/models") 

24@login_required 

25@exchange_required 

26def get_models(store_id: str): 

27 """ 

28 Retrieves models associated with a given store ID. 

29 

30 Args: 

31 store_id (int): The ID of the store for which to retrieve models. 

32 

33 Returns: 

34 dict: A dictionary containing the response message, success status, 

35 and data. 

36 """ 

37 

38 res = {} 

39 try: 

40 models = publisher_gateway.get_store_models(flask.session, store_id) 

41 res["success"] = True 

42 res["data"] = models 

43 response = make_response(res, 200) 

44 response.cache_control.max_age = "3600" 

45 except StoreApiResponseErrorList as error_list: 

46 error_messages = [ 

47 f"{error.get('message', 'An error occurred')}" 

48 for error in error_list.errors 

49 ] 

50 if "unauthorized" in error_messages: 

51 res["message"] = "Store not found" 

52 else: 

53 res["message"] = " ".join(error_messages) 

54 res["success"] = False 

55 response = make_response(res, 500) 

56 

57 return response 

58 

59 

60@models.route("/api/store/<store_id>/models/<model_name>/policies") 

61@login_required 

62@exchange_required 

63def get_policies(store_id: str, model_name: str): 

64 """ 

65 Get the policies for a given store model. 

66 

67 Args: 

68 store_id (str): The ID of the store. 

69 model_name (str): The name of the model. 

70 

71 Returns: 

72 dict: A dictionary containing the response message and success 

73 """ 

74 brand_id = get_brand_id(flask.session, store_id) 

75 res = {} 

76 

77 try: 

78 policies = publisher_gateway.get_store_model_policies( 

79 flask.session, brand_id, model_name 

80 ) 

81 res["success"] = True 

82 res["data"] = policies 

83 response = make_response(res, 200) 

84 response.cache_control.max_age = "3600" 

85 return response 

86 except StoreApiResponseErrorList as error_list: 

87 res["success"] = False 

88 res["message"] = " ".join( 

89 [ 

90 f"{error.get('message', 'An error occurred')}" 

91 for error in error_list.errors 

92 ] 

93 ) 

94 except Exception: 

95 res["success"] = False 

96 res["message"] = "An error occurred" 

97 

98 return make_response(res, 500) 

99 

100 

101@models.route( 

102 "/api/store/<store_id>/models/<model_name>/policies", methods=["POST"] 

103) 

104@login_required 

105@exchange_required 

106def create_policy(store_id: str, model_name: str): 

107 """ 

108 Creat policy for a store model. 

109 

110 Args: 

111 store_id (str): The ID of the store. 

112 model_name (str): The name of the model. 

113 

114 Returns: 

115 dict: A dictionary containing the response message and success 

116 """ 

117 signing_key = flask.request.form.get("signing_key") 

118 res = {} 

119 try: 

120 signing_keys_data = publisher_gateway.get_store_signing_keys( 

121 flask.session, store_id 

122 ) 

123 signing_keys = [key["sha3-384"] for key in signing_keys_data] 

124 

125 if not signing_key: 

126 res["message"] = "Signing key required" 

127 res["success"] = False 

128 return make_response(res, 500) 

129 

130 if signing_key in signing_keys: 

131 publisher_gateway.create_store_model_policy( 

132 flask.session, store_id, model_name, signing_key 

133 ) 

134 res["success"] = True 

135 else: 

136 res["message"] = "Invalid signing key" 

137 res["success"] = False 

138 except StoreApiResponseErrorList as error_list: 

139 res["success"] = False 

140 res["message"] = error_list.errors[0]["message"] 

141 

142 if res["success"]: 

143 return make_response(res, 200) 

144 return make_response(res, 500) 

145 

146 

147@models.route( 

148 "/api/store/<store_id>/models/<model_name>/policies/<revision>", 

149 methods=["DELETE"], 

150) 

151@login_required 

152@exchange_required 

153def delete_policy(store_id: str, model_name: str, revision: str): 

154 res = {} 

155 try: 

156 response = publisher_gateway.delete_store_model_policy( 

157 flask.session, store_id, model_name, revision 

158 ) 

159 if response.status_code == 204: 

160 res = {"success": True} 

161 return make_response(res, 200) 

162 elif response.status_code == 404: 

163 res = {"success": False, "message": "Policy not found"} 

164 return make_response(res, 404) 

165 except StoreApiResponseErrorList as error_list: 

166 res["success"] = False 

167 res["message"] = error_list.errors[0]["message"] 

168 return make_response(res, 500) 

169 

170 

171@models.route("/api/store/<store_id>/models", methods=["POST"]) 

172@login_required 

173@exchange_required 

174def create_models(store_id: str): 

175 """ 

176 Create a model for a given store. 

177 

178 Args: 

179 store_id (str): The ID of the store. 

180 

181 Returns: 

182 dict: A dictionary containing the response message and success 

183 status. 

184 """ 

185 

186 # TO DO: Addn validation that name does not exist already 

187 

188 res = {} 

189 

190 try: 

191 name = flask.request.form.get("name") 

192 api_key = flask.request.form.get("api_key", "") 

193 

194 if len(name) > 128: 

195 res["message"] = "Name is too long. Limit 128 characters" 

196 res["success"] = False 

197 return make_response(res, 500) 

198 

199 if api_key and len(api_key) != 50 and not api_key.isalpha(): 

200 res["message"] = "Invalid API key" 

201 res["success"] = False 

202 return make_response(res, 500) 

203 

204 publisher_gateway.create_store_model( 

205 flask.session, store_id, name, api_key 

206 ) 

207 

208 res["success"] = True 

209 return make_response(res, 201) 

210 except StoreApiResponseErrorList as error_list: 

211 res["success"] = False 

212 messages = [ 

213 f"{error.get('message', 'An error occurred')}" 

214 for error in error_list.errors 

215 ] 

216 res["message"] = (" ").join(messages) 

217 

218 except Exception: 

219 res["success"] = False 

220 res["message"] = "An error occurred" 

221 

222 return make_response(res, 500) 

223 

224 

225@models.route("/api/store/<store_id>/models/<model_name>", methods=["PATCH"]) 

226@login_required 

227@exchange_required 

228def update_model(store_id: str, model_name: str): 

229 """ 

230 Update a model for a given store. 

231 

232 Args: 

233 store_id (str): The ID of the store. 

234 model_name (str): The name of the model. 

235 

236 Returns: 

237 dict: A dictionary containing the response message and success 

238 status. 

239 """ 

240 res = {} 

241 

242 try: 

243 api_key = flask.request.form.get("api_key", "") 

244 

245 if len(api_key) != 50 and not api_key.isalpha(): 

246 res["message"] = "Invalid API key" 

247 res["success"] = False 

248 return make_response(res, 500) 

249 

250 publisher_gateway.update_store_model( 

251 flask.session, store_id, model_name, api_key 

252 ) 

253 res["success"] = True 

254 

255 except StoreApiResponseErrorList as error_list: 

256 res["success"] = False 

257 res["message"] = error_list.errors[0]["message"] 

258 

259 except StoreApiResourceNotFound: 

260 res["success"] = False 

261 res["message"] = "Model not found" 

262 if res["success"]: 

263 return make_response(res, 200) 

264 return make_response(res, 500)