Coverage for ghost/client.py: 84%

1import abc 

2import time 

3from dataclasses import dataclass, field 

4from datetime import datetime as dt 

5from json import JSONDecodeError 

6 

7import jwt 

8import requests 

9import yarl 

10 

11from .exceptions import * 

12from .resources import * 

13 

14MAX_ERROR_LIMIT = 3 

15 

16 

17@dataclass 

18class GhostClient(abc.ABC): 

19 url: str 

20 headers: dict = field(init=False, repr=False) 

21 contentAPIKey: str = field(repr=False) 

22 adminAPIKey: str = None 

23 api_version: str = "v4" # or v3 

24 

25 _session = requests.Session() 

26 

27 # noinspection PyUnreachableCode 

28 if False: 

29 # Types, instanciated with _setup_resources_on_self: 

30 # useful for IDE's 

31 # post: PostResource 

32 posts: PostResource = field(init=False, repr=False, compare=False) 

33 page: PageResource = field(init=False, repr=False, compare=False) 

34 pages: PageResource = field(init=False, repr=False, compare=False) 

35 author: AuthorResource = field(init=False, repr=False, compare=False) 

36 authors: AuthorResource = field(init=False, repr=False, compare=False) 

37 tag: TagResource = field(init=False, repr=False, compare=False) 

38 tags: TagResource = field(init=False, repr=False, compare=False) 

39 image: ImageResource = field(init=False, repr=False, compare=False) 

40 images: ImageResource = field(init=False, repr=False, compare=False) 

41 theme: ThemeResource = field(init=False, repr=False, compare=False) 

42 themes: ThemeResource = field(init=False, repr=False, compare=False) 

43 member: MemberResource = field(init=False, repr=False, compare=False) 

44 members: MemberResource = field(init=False, repr=False, compare=False) 

45 user: UserResource = field(init=False, repr=False, compare=False) 

46 users: UserResource = field(init=False, repr=False, compare=False) 

47 # End types 

48 

49 def _setup_resources_on_self(self, resources, content=False): 

50 """ 

51 Arguments: 

52 resources (list[type]) 

53 content (bool) 

54 """ 

55 for resource in resources: 

56 singular = resource.__name__.lower().split("resource")[0] 

57 

58 setattr(self, singular, resource(self, content=content, single=True)) 

59 plural = f"{singular}s" 

60 setattr(self, plural, resource(self, content=content)) 

61 

62 def _create_headers(self, api_version=None): 

63 """ 

64 Create the ghost authentication header 

65 

66 Args: 

67 api_version (string): for which API version to create a token 

68 

69 Returns: 

70 dict: Authorization header 

71 """ 

72 headers = {} 

73 

74 token = self._create_token(api_version) 

75 if token: 

76 headers["Authorization"] = f"Ghost {token}" 

77 

78 headers["accept-version"] = api_version 

79 

80 return headers 

81 

82 def _check_keys(self): 

83 raise NotImplementedError( 

84 "Implement _check_keys when inheriting from this class." 

85 ) 

86 

87 def _create_token(self, api_version: str = None): 

88 """ 

89 Create a JWT token if an admin API key was supplied. 

90 

91 Args: 

92 api_version (str): override the client's api version 

93 

94 Returns: 

95 str: auth token for ghost 

96 """ 

97 if not self._check_keys(): 

98 raise ValueError("Please enter valid auth keys!") 

99 

100 if self.adminAPIKey: 

101 if api_version is None: 

102 api_version = self.api_version 

103 

104 DURATION_IN_MINUTES = 5 

105 id, secret = self.adminAPIKey.split(":") 

106 iat = int(dt.now().timestamp()) 

107 header = {"alg": "HS256", "typ": "JWT", "kid": id} 

108 payload = { 

109 "iat": iat, 

110 "exp": iat + (DURATION_IN_MINUTES * 60), 

111 "aud": f"/{api_version}/admin/", 

112 } 

113 return jwt.encode( 

114 payload, bytes.fromhex(secret), algorithm="HS256", headers=header 

115 ) 

116 

117 def resource(self, name, single=False): 

118 """ 

119 Create an anonymous resource on the fly - to be used if there is no class available for some resource, 

120 that does have an endpoint in ghost. 

121 """ 

122 raise NotImplementedError("Implement these in the inherited classes.") 

123 

124 def _handle_errors(self, response: requests.Response): 

125 """ 

126 Raise custom ghost exceptions on different types of errors, 

127 instead of just returning the response JSON 

128 """ 

129 try: 

130 data = response.json() 

131 err = data.get("errors") 

132 if not err: 

133 raise GhostUnknownException( 

134 response.status_code, 

135 error_message="Unknown Error Occurred", 

136 exception=data, 

137 ) 

138 

139 main_error = err[0] 

140 raise GhostResponseException( 

141 response.status_code, 

142 main_error["type"], 

143 main_error["message"], 

144 exception=err, 

145 ) 

146 

147 except JSONDecodeError as e: 

148 raise GhostJSONException( 

149 response.status_code, error_message="JSON Parsing Failed", exception=e 

150 ) 

151 

152 def GET(self, url, params=None): 

153 """ 

154 Pass to self.interact with GET 

155 """ 

156 resp = self._interact("get", url, params=params) 

157 

158 if not resp.ok: 

159 self._handle_errors(resp) 

160 

161 return resp.json() 

162 

163 def DELETE(self, *_, **__): 

164 raise NotImplementedError("Implement this in the GhostAdmin class") 

165 

166 def PUT(self, *_, **__): 

167 raise NotImplementedError("Implement this in the GhostAdmin class") 

168 

169 def POST(self, *_, **__): 

170 raise NotImplementedError("Implement this in the GhostAdmin class") 

171 

172 def _interact( 

173 self, 

174 verb: str, 

175 endpoint: str, 

176 params: dict = None, 

177 files: dict = None, 

178 json: dict = None, 

179 api_version: str = None, 

180 ): 

181 """ 

182 Wrapper for requests that deals with Ghost API specifics and handles the response. 

183 

184 Args: 

185 verb (str): The HTTP verb to use. 

186 endpoint: The endpoint you want to access. 

187 For example, if you want to access the posts endpoint, you would pass in "posts". 

188 params (dict): A dictionary of query parameters to be appended to the URL. 

189 files (dict): a dictionary of files to upload. 

190 E.g. {"file": (name, file, mime_type)} 

191 json (dict): The JSON data to send in the body of the request. 

192 api_version (str): The version of the API you want to use. 

193 

194 Returns: 

195 requests.Response: A response object. 

196 """ 

197 if api_version is None: 

198 # default 

199 api_version = self.api_version 

200 headers = self.headers 

201 else: 

202 # custom api version, new headers: 

203 headers = self._create_headers(api_version) 

204 

205 verb = verb.lower() 

206 

207 # url + /ghost/api/ + /v3/admin/ + ... 

208 # in v5, api version is no longer sent in the URL 

209 

210 url = ( 

211 yarl.URL(self.url) 

212 / "ghost/api" 

213 / (api_version if api_version != "v5" else "") 

214 / endpoint 

215 ) 

216 

217 if endpoint.startswith("content") or not self.adminAPIKey: 

218 # yarl URL() % dict() encodes and adds query parameters as e.g. ?key=value 

219 url %= {"key": self.contentAPIKey} 

220 

221 error_count = 0 

222 

223 url = str(url) 

224 while error_count < MAX_ERROR_LIMIT: 

225 if verb == "get": 

226 resp = self._session.get(url, headers=self.headers, params=params) 

227 elif verb == "post": 

228 resp = self._session.post( 

229 url, headers=self.headers, params=params, files=files, json=json 

230 ) 

231 elif verb == "put": 

232 resp = self._session.put( 

233 url, headers=self.headers, params=params, files=files, json=json 

234 ) 

235 elif verb == "delete": 

236 resp = self._session.delete(url, headers=self.headers, params=params) 

237 else: 

238 raise ValueError(f"Unknown verb: {verb}") 

239 

240 if resp.status_code == 401 and not error_count: 

241 # retry instantly with new headers 

242 self.headers = self._create_headers() 

243 error_count += 1 

244 elif resp.status_code == 401 and error_count: 

245 # after the first error, try again after a timeout 

246 time.sleep(5) 

247 self.headers = self._create_headers() 

248 error_count += 1 

249 else: 

250 # on other error codes, print and return 

251 if not resp.ok: 

252 # print( 

253 # { 

254 # "endpoint": url, 

255 # "method": verb, 

256 # "code": resp.status_code, 

257 # "message": resp.text, 

258 # }, 

259 # file=sys.stderr, 

260 # ) 

261 pass 

262 return resp 

263 

264 raise IOError("Could not contact API correctly after 3 tries.") 

265 

266 

267@dataclass 

268class GhostContent(GhostClient): 

269 def _check_keys(self): 

270 """ 

271 This Client only requires a Content Key 

272 """ 

273 return self.contentAPIKey 

274 

275 def __post_init__(self): 

276 """ 

277 Set up the different Resources 

278 """ 

279 

280 self.headers = {} 

281 

282 # resources: 

283 self._setup_resources_on_self( 

284 [ 

285 PostResource, 

286 PageResource, 

287 AuthorResource, 

288 TagResource, 

289 ImageResource, 

290 ThemeResource, 

291 MemberResource, 

292 UserResource, 

293 ], 

294 content=True, 

295 ) 

296 

297 # there's only one site/one settings: 

298 self.site = SiteResource(self, single=True, content=True) 

299 self.settings = SettingsResource(self, single=True, content=True) 

300 

301 def DELETE(self, *_, **__): 

302 raise GhostWrongApiError("DELETE is not allowed for the content API!") 

303 

304 def PUT(self, *_, **__): 

305 raise GhostWrongApiError("PUT is not allowed for the content API!") 

306 

307 def POST(self, *_, **__): 

308 raise GhostWrongApiError("POST is not allowed for the content API!") 

309 

310 def resource(self, name, single=False): 

311 """ 

312 Create an anonymous resource on the fly - to be used if there is no class available for some resource, 

313 that does have an endpoint in ghost. 

314 """ 

315 

316 class _Resource(GhostContentResource): 

317 # Temporary Resource 

318 resource = name 

319 

320 return _Resource(self, single=single) 

321 

322 

323@dataclass 

324class GhostAdmin(GhostClient): 

325 url: str 

326 headers: dict = field(init=False, repr=False) 

327 contentAPIKey: str = field(repr=False) 

328 adminAPIKey: str = field(repr=False) 

329 api_version: str = "v4" # or v3 or v5 

330 

331 _session = requests.Session() 

332 

333 def _check_keys(self): 

334 """ 

335 The admin API requires both an admin api key and a content api key 

336 """ 

337 return self.adminAPIKey and self.contentAPIKey 

338 

339 def __post_init__(self): 

340 """ 

341 Setup the JWT Authentication headers and the different Resources 

342 """ 

343 

344 self.headers = self._create_headers() 

345 

346 # resources: 

347 self._setup_resources_on_self( 

348 [ 

349 PostResource, 

350 PageResource, 

351 AuthorResource, 

352 TagResource, 

353 ImageResource, 

354 ThemeResource, 

355 MemberResource, 

356 UserResource, 

357 ], 

358 content=False, 

359 ) 

360 

361 # there's only one site/one settings: 

362 self.site = SiteResource(self, single=True) 

363 self.settings = SettingsResource(self, single=True) 

364 

365 def POST(self, url, params=None, json=None, files=None): 

366 """ 

367 Pass to self.interact with POST 

368 

369 Returns: 

370 dict: response data 

371 """ 

372 resp = self._interact("post", url, params=params, json=json, files=files) 

373 

374 if not resp.ok: 

375 self._handle_errors(resp) 

376 

377 return resp.json() 

378 

379 def PUT(self, url, params=None, json=None, files=None): 

380 """ 

381 Pass to self.interact with PUT 

382 

383 Returns: 

384 dict: response data 

385 """ 

386 resp = self._interact("put", url, params=params, json=json) 

387 

388 if not resp.ok: 

389 self._handle_errors(resp) 

390 

391 return resp.json() 

392 

393 def DELETE(self, url, params=None): 

394 """ 

395 Pass to self.interact with DELETE 

396 

397 Returns: 

398 bool: if the status code is right 

399 """ 

400 return self._interact("delete", url, params=params).status_code == 204 

401 

402 def resource(self, name, single=False): 

403 """ 

404 Create an anonymous resource on the fly - to be used if there is no class available for some resource, 

405 that does have an endpoint in ghost. 

406 """ 

407 

408 class _Resource(GhostAdminResource): 

409 # Temporary Resource 

410 resource = name 

411 

412 return _Resource(self, single=single)