Coverage for portality/api/common.py: 50%

160 statements  

« prev     ^ index     » next       coverage.py v6.4.2, created at 2022-07-20 16:12 +0100

1#~~API:Feature~~ 

2import json, uuid 

3from portality.core import app 

4from flask import request 

5from copy import deepcopy 

6from link_header import LinkHeader, Link 

7 

8LINK_HEADERS = ['next', 'prev', 'last'] 

9TOTAL_RESULTS_COUNT = ['total'] 

10 

11ERROR_TEMPLATE = {"status": {"type": "string"}, "error": {"type": "string"}} 

12CREATED_TEMPLATE = {"status": {"type": "string"}, "id": {"type": "string"}, "location": {"type": "string"}} 

13 

14 

15class Api(object): 

16 # ~~->Swagger:Feature~~ 

17 # ~~->API:Documentation~~ 

18 SWAG_TEMPLATE = { 

19 "description" : "", 

20 "responses": {}, 

21 "parameters": [], 

22 "tags": [] 

23 } 

24 R200 = {"schema": {}, "description": "A successful request/response"} 

25 R201 = {"schema": {"properties": CREATED_TEMPLATE}, "description": "Resource created successfully, response " 

26 "contains the new resource ID and location."} 

27 R201_BULK = {"schema": {"items": {"properties" : CREATED_TEMPLATE, "type" : "object"}, "type" : "array"}, 

28 "description": "Resources created successfully, response contains the new resource IDs " 

29 "and locations."} 

30 R204 = {"description": "OK (Request succeeded), No Content"} 

31 R400 = {"schema": {"properties": ERROR_TEMPLATE}, "description": "Bad Request. Your request body was missing a " 

32 "required field, or the data in one of the " 

33 "fields did not match the schema above (e.g. " 

34 "string of latin letters in an integer field). " 

35 "In the Bulk API it may mean that one of the " 

36 "records in the bulk operation failed. See the " 

37 "\"error\" part of the response for details."} 

38 R401 = {"schema": {"properties": ERROR_TEMPLATE}, "description": "Access to this route/resource requires " 

39 "authentication, but you did not provide any " 

40 "credentials."} 

41 R403 = {"schema": {"properties": ERROR_TEMPLATE}, "description": "Access to this route/resource requires " 

42 "authentication, and you provided the wrong " 

43 "credentials. This includes situations where you " 

44 "are authenticated successfully via your API " 

45 "key, but you are not the owner of a specific " 

46 "resource and are therefore barred from " 

47 "updating/deleting it."} 

48 R404 = {"schema": {"properties": ERROR_TEMPLATE}, "description": "Resource not found"} 

49 R409 = {"schema": {"properties": ERROR_TEMPLATE}, "description": "This resource or one it depends on is currently " 

50 "locked for editing by another user, and you may " 

51 "not submit changes to it at this time"} 

52 R500 = {"schema": {"properties": ERROR_TEMPLATE}, "description": "Unable to retrieve the recource. This record " 

53 "contains bad data"} 

54 

55 SWAG_API_KEY_REQ_PARAM = { 

56 "description": "<div class=\"search-query-docs\">Go to 'MY ACCOUNT' and 'Settings' to find your API key. If there is no key, click 'Generate a new API key'. If you do not see that button, contact us.", 

57 "required": True, 

58 "type": "string", 

59 "name": "api_key", 

60 "in": "query" 

61 } 

62 

63 @classmethod 

64 @property 

65 def SWAG_TAG(cls): 

66 raise RuntimeError('You must override this class constant in every subclass.') 

67 

68 @classmethod 

69 def _add_swag_tag(cls, template): 

70 template['tags'].append(cls.SWAG_TAG) 

71 return template 

72 

73 @classmethod 

74 def _add_api_key(cls, template, optional=False): 

75 # ~~->APIKey:Feature~~ 

76 api_key_param = deepcopy(cls.SWAG_API_KEY_REQ_PARAM) 

77 if optional: 

78 api_key_param['required'] = False 

79 api_key_param['description'] = "<div class=\"search-query-docs\"><em>Note this parameter is optional for " \ 

80 "this route - you could, but don't have to supply a key. Doing so grants " \ 

81 "you access to records of yours that are not public, in addition to all " \ 

82 "public records.</em> Go to 'MY ACCOUNT' and 'Settings' to find your API key. If there is no key, click 'Generate a new API key'. If you do not see that button, contact us." 

83 template["parameters"].insert(0, api_key_param) 

84 return template 

85 

86 @classmethod 

87 def _build_swag_response(cls, template, api_key_optional_override=None, api_key_override=None): 

88 """ 

89 Construct the swagger response structure upon a template 

90 :param template 

91 :param api_key_optional_override: override the class-level value of API_KEY_OPTIONAL 

92 :return: an updated template 

93 """ 

94 template = deepcopy(template) 

95 cls._add_swag_tag(template) 

96 if api_key_override is not False: 

97 if api_key_optional_override is not None: 

98 cls._add_api_key(template, optional=api_key_optional_override) 

99 elif hasattr(cls, 'API_KEY_OPTIONAL'): 

100 cls._add_api_key(template, optional=cls.API_KEY_OPTIONAL) 

101 

102 return template 

103 

104 

105class Api400Error(Exception): 

106 pass 

107 

108 

109class Api401Error(Exception): 

110 pass 

111 

112 

113class Api403Error(Exception): 

114 pass 

115 

116 

117class Api404Error(Exception): 

118 pass 

119 

120 

121class Api409Error(Exception): 

122 """ 

123 API error to throw if a resource being edited is locked 

124 """ 

125 pass 

126 

127 

128class Api500Error(Exception): 

129 pass 

130 

131 

132class DataObjectJsonEncoder(json.JSONEncoder): 

133 def default(self, o): 

134 return o.data 

135 

136 

137class ModelJsonEncoder(json.JSONEncoder): 

138 def default(self, o): 

139 return o.data 

140 

141 

142def created(obj, location): 

143 app.logger.info("Sending 201 Created: {x}".format(x=location)) 

144 t = deepcopy(CREATED_TEMPLATE) 

145 t['status'] = "created" 

146 t['id'] = obj.id 

147 t['location'] = location 

148 resp = respond(json.dumps(t), 201) 

149 resp.headers["Location"] = location 

150 resp.status_code = 201 

151 return resp 

152 

153 

154def bulk_created(ids_and_locations): 

155 app.logger.info("Sending 201 Created for bulk request") 

156 out = [] 

157 for id, loc in ids_and_locations: 

158 t = deepcopy(CREATED_TEMPLATE) 

159 t['status'] = "created" 

160 t['id'] = id 

161 t['location'] = loc 

162 out.append(t) 

163 

164 resp = respond(json.dumps(out), 201) 

165 resp.status_code = 201 

166 return resp 

167 

168 

169def no_content(): 

170 return respond("", 204) 

171 

172 

173def jsonify_data_object(do): 

174 data = json.dumps(do, cls=DataObjectJsonEncoder) 

175 return respond(data, 200) 

176 

177 

178def jsonify_models(models): 

179 data = json.dumps(models, cls=ModelJsonEncoder) 

180 

181 metadata = {} 

182 for k in LINK_HEADERS + TOTAL_RESULTS_COUNT: 

183 if k in models.data: 

184 metadata[k] = models.data[k] 

185 

186 return respond(data, 200, metadata=metadata) 

187 

188 

189def generate_link_headers(metadata): 

190 """ 

191 Generate Link: HTTP headers for API navigation. 

192 

193 :param metadata: Dictionary with none, some or all of the 

194 keys 'next', 'prev' and 'last' defined. The values are the 

195 corresponding pre-generated links. 

196 """ 

197 link_metadata = {k: v for k, v in metadata.items() if k in LINK_HEADERS} 

198 

199 links = [] 

200 for k, v in link_metadata.items(): 

201 links.append(Link(v, rel=k)) # e.g. Link("https://example.com/foo", rel="next") 

202 

203 return str(LinkHeader(links)) # RFC compliant headers e.g. 

204 # <https://example.com/foo>; rel=next, <https://example.com/bar>; rel=last 

205 

206 

207def respond(data, status, metadata=None): 

208 # avoid subtle bugs, don't use mutable objects as default vals in Python 

209 # https://pythonconquerstheuniverse.wordpress.com/category/python-gotchas/ 

210 if metadata is None: 

211 metadata = {} 

212 

213 headers = {'Access-Control-Allow-Origin': '*'} 

214 link = generate_link_headers(metadata) 

215 if link: 

216 headers['Link'] = link 

217 

218 if 'total' in metadata: 

219 headers['X-Total-Count'] = metadata['total'] 

220 

221 callback = request.args.get('callback', False) 

222 if callback: 

223 content = str(callback) + '(' + str(data) + ')' 

224 return app.response_class(content, status, headers, mimetype='application/javascript') 

225 else: 

226 return app.response_class(data, status, headers, mimetype='application/json') 

227 

228 

229@app.errorhandler(Api400Error) 

230def bad_request(error): 

231 magic = uuid.uuid1() 

232 app.logger.info("Sending 400 Bad Request from client: {x} (ref: {y})".format(x=str(error), y=magic)) 

233 t = deepcopy(ERROR_TEMPLATE) 

234 t['status'] = 'bad_request' 

235 t['error'] = str(error) + " (ref: {y})".format(y=magic) 

236 return respond(json.dumps(t), 400) 

237 

238 

239@app.errorhandler(Api404Error) 

240def not_found(error): 

241 magic = uuid.uuid1() 

242 app.logger.info("Sending 404 Not Found from client: {x} (ref: {y})".format(x=str(error), y=magic)) 

243 t = deepcopy(ERROR_TEMPLATE) 

244 t['status'] = 'not_found' 

245 t['error'] = str(error) + " (ref: {y})".format(y=magic) 

246 return respond(json.dumps(t), 404) 

247 

248 

249@app.errorhandler(Api401Error) 

250def unauthorised(error): 

251 magic = uuid.uuid1() 

252 app.logger.info("Sending 401 Unauthorised from client: {x} (ref: {y})".format(x=str(error), y=magic)) 

253 t = deepcopy(ERROR_TEMPLATE) 

254 t['status'] = 'unauthorised' 

255 t['error'] = str(error) + " (ref: {y})".format(y=magic) 

256 return respond(json.dumps(t), 401) 

257 

258 

259@app.errorhandler(Api403Error) 

260def forbidden(error): 

261 magic = uuid.uuid1() 

262 app.logger.info("Sending 403 Forbidden from client: {x} (ref: {y})".format(x=str(error), y=magic)) 

263 t = deepcopy(ERROR_TEMPLATE) 

264 t['status'] = 'forbidden' 

265 t['error'] = str(error) + " (ref: {y})".format(y=magic) 

266 return respond(json.dumps(t), 403) 

267 

268 

269@app.errorhandler(Api500Error) 

270def bad_request(error): 

271 magic = uuid.uuid1() 

272 app.logger.info("Sending 500 Bad Request from client: {x} (ref: {y})".format(x=str(error), y=magic)) 

273 t = deepcopy(ERROR_TEMPLATE) 

274 t['status'] = 'Unable to retrieve the recource.' 

275 t['error'] = str(error) + " (ref: {y})".format(y=magic) 

276 return respond(json.dumps(t), 500)