Coverage for ghost/results.py: 92%

1from __future__ import annotations 

2 

3import hashlib 

4import json 

5 

6# noinspection PyUnreachableCode 

7if False: 

8 # annotation only 

9 from .abs_resources import GhostResource 

10 

11 

12def dict_hash(dictionary: dict): 

13 """ 

14 MD5 hash of a dictionary, used to compare dicts 

15 """ 

16 dhash = hashlib.md5() 

17 # We need to sort arguments so {'a': 1, 'b': 2} is 

18 # the same as {'b': 2, 'a': 1} 

19 encoded = json.dumps(dictionary, sort_keys=True).encode() 

20 dhash.update(encoded) 

21 return dhash.hexdigest() 

22 

23 

24def is_admin_resource(obj): 

25 """ 

26 GhostAdminResource cannot be imported globally due to circular referencing. 

27 This method checks if an object is a Ghost Admin Resource 

28 """ 

29 from .abs_resources import GhostAdminResource 

30 

31 return isinstance(obj, GhostAdminResource) 

32 

33 

34class GhostResult: 

35 """ 

36 Single resource value (e.g. one post, one tag) 

37 """ 

38 

39 def __init__(self, d: dict, resource: GhostResource): 

40 if d.get("tags") and isinstance(d["tags"], list): 

41 d["tags"] = {_["slug"]: _ for _ in d["tags"]} # list to dict 

42 self.__data__ = d 

43 self._resource: GhostResource = resource 

44 

45 def __getitem__(self, item): 

46 """ 

47 Allows to access the result as a dict (result[item]) 

48 """ 

49 return self.__data__.get(item) 

50 

51 def __getattr__(self, item): 

52 """ 

53 Allows to access the result as an object (result.item) 

54 """ 

55 return self.__getitem__(item) 

56 

57 def __repr__(self): 

58 """ 

59 human-friendlier representation of the resource 

60 """ 

61 return f"<{self._resource.resource}: {self.slug}>" 

62 

63 def as_dict(self): 

64 """ 

65 Get the Result's raw values as dict 

66 """ 

67 return self.__data__ 

68 

69 def delete(self): 

70 """ 

71 Delete this response's item 

72 """ 

73 rs = self._resource 

74 if not is_admin_resource(rs): 

75 raise PermissionError("Please use the admin API to delete resources.") 

76 return rs.delete(self.id) 

77 

78 def update(self, **data): 

79 """ 

80 Update this response's item 

81 """ 

82 rs = self._resource 

83 if not is_admin_resource(rs): 

84 raise PermissionError("Please use the admin API to update resources.") 

85 return rs.update(self.id, data, self) 

86 

87 def __eq__(self, other): 

88 """ 

89 Check if this result's data matches the other's 

90 """ 

91 return dict_hash(self.__data__) == dict_hash(other.__data__) 

92 

93 

94class GhostResultSet: 

95 """ 

96 List of resource objects (e.g. posts, tags) 

97 """ 

98 

99 def __init__(self, lst: list, resource: GhostResource, meta: dict, request: dict): 

100 self.__list__ = [ 

101 (_ if isinstance(_, GhostResult) else GhostResult(_, resource)) for _ in lst 

102 ] 

103 self._resource = resource 

104 meta["request"] = request 

105 self._meta = meta 

106 

107 def __repr__(self): 

108 return f"[{', '.join([repr(_) for _ in self.__list__])}]" 

109 

110 def __iter__(self): 

111 """ 

112 Iterate the result list 

113 """ 

114 for _ in self.__list__: 

115 yield _ 

116 

117 def __len__(self): 

118 return len(self.__list__) 

119 

120 def __getitem__(self, idx): 

121 """ 

122 Get an item by index (resultset[idx]) 

123 """ 

124 return self.__list__[idx] 

125 

126 def __or__(self, other: GhostResultSet): 

127 """ 

128 Set | Set -> Bigger Set 

129 

130 Note: some metadata will be lost since other._meta is not returned in the new set 

131 """ 

132 if self._resource != other._resource: 

133 raise TypeError("Can only combine ResultSets of the same Resource Type") 

134 

135 combined_list = [*self.__list__, *other.__list__] 

136 

137 return GhostResultSet( 

138 combined_list, self._resource, self._meta, self._meta["request"] 

139 ) 

140 

141 def as_dict(self): 

142 return {_["id"]: _.as_dict() for _ in self.__list__} 

143 

144 def as_list(self): 

145 return [_.as_dict() for _ in self.__list__] 

146 

147 def delete(self): 

148 """ 

149 Delete all results in this set 

150 """ 

151 return [i.delete() for i in self.__list__] 

152 

153 def update(self, **data): 

154 """ 

155 Update all results in this set 

156 """ 

157 return [i.update(**data) for i in self.__list__] 

158 

159 def next(self): 

160 """ 

161 Get the next page for a resultset 

162 """ 

163 pag = self._meta["pagination"] 

164 request = self._meta["request"] 

165 params = request["params"] 

166 params["limit"] = pag["limit"] 

167 params["page"] = pag["next"] 

168 if not params["page"]: 

169 return [] 

170 

171 # request contains path, params, single 

172 # used to keep the request params the same (so pagination makes sense) 

173 # except the page number (and force limit received from server to be sure) 

174 return self._resource._get(**request)