Coverage for ghost/abs_resources.py: 83%
179 statements
« prev ^ index » next coverage.py v7.5.3, created at 2024-06-17 17:19 +0200
1from __future__ import annotations # allow type hint without actual import
3import abc
4import json
5from abc import ABC
6from typing import Iterable
8# noinspection PyUnreachableCode
9if False:
10 # prevent circular import
11 from .client import GhostClient
13from .exceptions import GhostResourceNotFoundException
14from .results import GhostResult, GhostResultSet
17def is_iterable(value):
18 """
19 "If the value is iterable and not a string, return True."
21 The first part of the function checks if the value is iterable. The second part checks if the value is not a string
23 Args:
24 value: The value to check.
26 Returns:
27 A boolean value.
28 """
29 return isinstance(value, Iterable) and not isinstance(value, str)
32class GhostResource(abc.ABC):
33 single: bool
34 content: bool
35 ga: GhostClient
37 def __repr__(self):
38 return f"<GhostResource {self.resource}>"
40 @property
41 def resource(self):
42 raise NotImplementedError("Please choose a resource")
44 @property
45 def api(self):
46 raise NotImplementedError("Please choose an api")
48 def __init__(self, ghost_admin: GhostClient, single=False, content=False):
49 self.ga = ghost_admin
50 self.single = single
51 self.content = content
53 # def __call__(self, id: str = None, /, **filters): # <- Python 3.8+
54 def __call__(self, id: str = None, **filters):
55 """
56 Magic method to make it possible to call something like ghost.pages(tag='sometag') or ghost.page('some id')
57 instead of ghots.pages.get(tag='sometag')
58 """
60 return self.get(id, **filters)
62 def _list_join(self, value: list, paren: str = "square"):
63 """
64 Join a list and wrap it in different parentheses
65 """
67 value = ",".join(value)
68 if not paren:
69 return value
70 elif paren == "square":
71 return f"[{value}]"
72 elif paren == "round":
73 return f"({value})"
74 else:
75 # todo: other?
76 raise NotImplementedError(f"Parentheses type '{paren}' not supported.")
78 def _filters_to_ghost(self, filters: dict):
79 """
80 Ghost requires params to be supplied as strings,
81 but for developer experience it is much nicer to work with e.g. lists.
82 This method converts this for ease of use.
85 Returns:
86 str: filters joined by +
87 """
88 ghost_filters = []
90 for key, value in filters.items():
91 if is_iterable(value):
92 value = self._list_join(value)
94 ghost_filters.append(f"{key}:{value}")
96 return "+".join(ghost_filters)
98 def _create_url(self, path: Iterable):
99 """
100 Build a valid Ghost API endpoint URL.
102 Args:
103 path (Iterable): parts of path to combine
105 Returns:
106 str: URL path, joined by /
107 """
108 api = "content" if self.content else self.api
109 return "/".join([api, self.resource, *path])
111 def GET(self, *path, **args):
112 """
113 Perform a GET request
114 (forwarded to the Client passed to this class)
116 Returns:
117 GhostResult | GhostResultSet: depending on the request
118 """
119 url = self._create_url(path)
120 return self.ga.GET(url, **args)
122 def POST(self, *path, **args):
123 """
124 Perform a POST request
125 (forwarded to the Client passed to this class)
127 Returns:
128 dict: json response
129 """
130 url = self._create_url(path)
131 return self.ga.POST(url, **args)
133 def PUT(self, *path, **args):
134 """
135 Perform a PUT request
136 (forwarded to the Client passed to this class)
138 Returns:
139 dict: json reponse
140 """
141 url = self._create_url(path)
142 return self.ga.PUT(url, **args)
144 def DELETE(self, *path, **args):
145 """
146 Perform a DELETE request
147 (forwarded to the Client passed to this class)
149 Returns:
150 bool: if successfull
151 """
152 url = self._create_url(path)
153 return self.ga.DELETE(url, **args)
155 def _create_args(self, d: dict):
156 """
157 Turn arguments such as fields, page, order, filters etc. into the format Ghost expects
158 """
160 # todo: use Operators (greater than etc), combinations (+ for AND), etc.
161 d.pop("self") # needed since locals() is passed to this method
162 args = {}
164 for key, value in d.items():
165 if value is None:
166 continue
167 if isinstance(value, dict):
168 value = self._filters_to_ghost(value)
169 elif is_iterable(value):
170 value = self._list_join(value, paren="")
172 args[key] = value
174 return args
176 def _get(self, path: str = "", params: dict = None, single: bool = None):
177 """
178 GET some resource and handle the result(s)
180 Args:
181 path (str): The path to the resource you want to get.
182 params (dict): A dictionary of query parameters to be passed to the API.
183 single (bool): If True, a GhostResult will be returned, even when the Resource is not single.
184 Otherwise, a GhostResultSet will be returned.
186 Returns:
187 GhostResult | GhostResultSet
188 """
190 if params is None:
191 params = {}
193 resp = self.GET(path, params=params)
194 data = resp.get(self.resource)
196 if not data and self.single:
197 raise GhostResourceNotFoundException(200, "Resource Not Found", path)
199 if self.single or single:
200 return GhostResult(data[0] if isinstance(data, list) else data, self)
201 else:
202 return GhostResultSet(
203 data,
204 self,
205 meta=resp["meta"],
206 request={
207 "path": path,
208 "params": params,
209 "single": single,
210 },
211 )
213 def paginate(self, *, per: int = 25, **filters):
214 """
215 Generator that yields all the data for this resource
217 Args:
218 per (int): The number of results to return per page. Defaults to 25
219 filters: modifiers passed to the GET request
221 Yields:
222 GhostResult: items matching the supplied filters
223 """
225 data = True
226 page = 1
227 filters["limit"] = per
229 while data:
230 filters["page"] = page
231 try:
232 data = self.get(**filters)
233 for d in data:
234 yield d
235 except GhostResourceNotFoundException:
236 break
237 page += 1
239 def _get_by_id(self, id: str, **_params):
240 """
241 Get a specific instance of this resource, by ID.
243 Args:
244 id (str): The id of the item to retrieve.
245 params: modifiers such as 'fields' to limit which columns to get
247 Returns:
248 GhostResult: item with id
249 """
251 params = {"formats": "html,mobiledoc", **_params}
252 return self._get(id, params, single=True)
254 def _get_by_filters(
255 self,
256 limit: int = None,
257 page: int = None,
258 order: str = None,
259 fields: list = None,
260 **filter,
261 ):
262 """
263 Get resource items matching filter
265 Args:
266 limit (int): The number of results to return.
267 page (int): The page number of the results to return.
268 order (str): The order in which to return the posts.
269 fields (list): A list of fields to include in the response.
270 filter: a dictionary of key-value pairs to filter by (e.g. author, slug, etc.)
272 Returns:
273 GhostResultSet:
274 """
275 args = self._create_args(locals())
276 args["formats"] = "html,mobiledoc"
278 return self._get(params=args)
280 # def get(self, id: str = None, /, **filters): # <- Python 3.8+
281 def get(self, id: str = None, **filters):
282 """
283 Either get
285 Args:
286 id (str): if the ID of the item is known
287 filters: parameters to filter data on.
288 See https://ghost.org/docs/admin-api/#parameters and https://ghost.org/docs/content-api/#parameters
289 for more info
291 Returns:
292 GhostResult | GhostResultSet: depending on if 'single' is used.
293 """
295 if id is not None:
296 return self._get_by_id(id)
297 elif "slug" in filters and len(filters) == 1:
298 return self._get_by_id(f"slug/{filters['slug']}")
299 else:
300 if "fields" in filters:
301 if "id" not in filters["fields"]:
302 filters["fields"].append("id")
304 if "updated_at" not in filters["fields"]:
305 filters["fields"].append("updated_at")
307 return self._get_by_filters(**filters)
309 def delete(self, *_, **__):
310 raise NotImplementedError("Implement this in the Admin Resources")
312 def update(self, *_, **__):
313 raise NotImplementedError("Implement this in the Admin Resources")
315 def create(self, *_, **__):
316 raise NotImplementedError("Implement this in the Admin Resources")
319class GhostAdminResource(GhostResource, ABC):
320 api = "admin"
322 def __md_card(self, md: str, idx: int = 0):
323 """
324 Generate a mobiledoc card for markdown
326 Args:
327 md (str): markdown text
328 idx (int): index, used to give cards a unique cardName
330 Returns:
331 list: mobiledoc formatted card
332 """
333 return [
334 "markdown",
335 {"cardName": f"markdown-{idx}", "markdown": md},
336 ]
338 def _transform_markdown(self, item: dict):
339 """
340 Allow developers to use markdown with mobiledoc more easily.
341 Currently not implemented as only one block of markdown shows up in Ghost instead of everything.
343 Args:
344 item (dict): the resource to be created. The value of markdown in item can be a string or a list of strings.
346 Returns:
347 None: this method only edits item
348 """
349 if item.get("markdown"):
350 raise NotImplementedError(
351 "Creating posts with markdown is currently not yet supported."
352 )
353 md = item["markdown"]
355 if is_iterable(md):
356 cards = [self.__md_card(_, i) for i, _ in enumerate(md)]
357 else:
358 cards = [self.__md_card(md)]
360 item["mobiledoc"] = json.dumps(
361 {
362 "version": "0.3.1",
363 "markups": [],
364 "atoms": [],
365 "cards": cards,
366 "sections": [[10, 0]],
367 }
368 )
370 del item["markdown"]
372 def _create_multiple(self, items: list):
373 raise NotImplementedError("Can only create one item at a time!")
374 data = {self.resource: items}
376 return self.POST(json=data)
378 def _create_one(self, item: dict):
379 """
380 Wrapper to create a new item of this resource
381 """
383 self._transform_markdown(item)
385 data = {self.resource: [item]}
387 if item.get("html"):
388 params = {"source": "html"}
389 # elif item.get('mobiledoc'):
390 # params = {"source": "mobiledoc"}
391 else:
392 params = {}
394 return self.POST(params=params, json=data)
396 def create(self, *a, **kw):
397 """
398 Create one or more new items.
399 One item is created if kwargs are used:
400 e.g. ghost.posts.create(title="something")
402 Multiple items are created if args is used:
403 e.g. ghost.posts.create({...}, {...})
406 """
408 if a and kw:
409 raise ValueError(
410 "Please use either only arguments or only keyword arguments."
411 )
412 elif a:
413 # return self._create_multiple(a)
414 return [self._create_one(_) for _ in a]
415 else: # kw
416 # return self._create_multiple([kw])
417 return self._create_one(kw)
419 def _delete_by_id(self, id: str):
420 """
421 Delete a specific item by ID
422 """
423 return self.DELETE(id)
425 def _delete_by_filters(self, filters):
426 """
427 Find all items matching filters and delete these
429 Returns:
430 list[bool]: success of each delete
431 """
432 try:
433 if not filters.get("limit"):
434 filters["limit"] = "all"
436 ids = self._get_by_filters(**filters)
437 if not ids:
438 return []
440 return [self._delete_by_id(_["id"]) for _ in ids]
441 except GhostResourceNotFoundException:
442 return []
444 # def delete(self, id=None, /, **filters): # <- Python 3.8+
445 def delete(self, id=None, **filters):
446 """
447 Delete either one item if 'id' is supplied or all items matching filters
448 """
450 if id is not None:
451 return self._delete_by_id(id)
452 else:
453 filters["fields"] = "id" # not more is needed for delete
454 return self._delete_by_filters(filters=filters)
456 def _update_by_id(self, id: str, data: dict, old=None):
457 """
458 PUT new data for ID
460 Args:
461 id (str): item to update
462 data (dict): data to update
463 old (dict|GhostResult): required for the old updated_at
465 Returns:
466 dict: response data
467 """
468 if old is None:
469 old = self._get_by_id(id, fields=["updated_at"])
471 data["updated_at"] = old["updated_at"]
473 return self.PUT(id, json={self.resource: [data]})
475 def _update_by_filters(self, data: dict, filters: dict):
476 """
477 For each item matching filters, update its data
479 Returns:
480 list[dict]: result of each PUT
481 """
483 try:
484 if not filters.get("limit"):
485 filters["limit"] = "all"
487 ids = self._get_by_filters(**filters, fields=["id", "updated_at"])
488 return [self._update_by_id(old["id"], data, old) for old in ids]
489 except GhostResourceNotFoundException:
490 return []
492 # def update(self, id: str = None, data: dict = None, old=None, /, **filters): # <- Python 3.8+
493 def update(self, id: str = None, data: dict = None, old=None, **filters):
494 """
495 Update either one item if 'id' is supplied or all items matching filters.
497 Args:
498 id (str): one item to update
499 data (dict): new data
500 old (dict|GhostResult): the old updated at is required to update a post.
501 If old is not supplied, it will be GET-requested before updating.
502 filters: to find items
504 Returns:
505 dict | list[dict]: response(s) from Ghost
506 """
508 if data is None:
509 raise ValueError("Please include new and old values in data!")
511 if id is not None:
512 return self._update_by_id(id, data=data, old=old)
513 else:
514 return self._update_by_filters(data=data, filters=filters)
517class GhostContentResource(GhostResource, ABC):
518 api = "content"