Coverage for mindsdb / integrations / handlers / confluence_handler / confluence_api_client.py: 92%

1from typing import List 

2 

3import requests 

4 

5 

6class ConfluenceAPIClient: 

7 def __init__(self, url: str, username: str, password: str): 

8 self.url = url 

9 self.username = username 

10 self.password = password 

11 self.session = requests.Session() 

12 self.session.auth = (self.username, self.password) 

13 self.session.headers.update({"Accept": "application/json"}) 

14 

15 def get_spaces( 

16 self, 

17 ids: List[int] = None, 

18 keys: List[str] = None, 

19 space_type: str = None, 

20 status: str = None, 

21 sort_condition: str = None, 

22 limit: int = None, 

23 ): 

24 url = f"{self.url}/wiki/api/v2/spaces" 

25 params = { 

26 "description-format": "view", 

27 } 

28 if ids: 

29 params["ids"] = ids 

30 if keys: 

31 params["keys"] = keys 

32 if space_type: 

33 params["type"] = space_type 

34 if status: 

35 params["status"] = status 

36 if sort_condition: 36 ↛ 37line 36 didn't jump to line 37 because the condition on line 36 was never true

37 params["sort"] = sort_condition 

38 if limit: 

39 params["limit"] = limit 

40 

41 return self._paginate(url, params) 

42 

43 def get_pages( 

44 self, 

45 page_ids: List[int] = None, 

46 space_ids: List[int] = None, 

47 statuses: List[str] = None, 

48 title: str = None, 

49 sort_condition: str = None, 

50 limit: int = None, 

51 ) -> List[dict]: 

52 url = f"{self.url}/wiki/api/v2/pages" 

53 params = { 

54 "body-format": "storage", 

55 } 

56 if page_ids: 

57 params["id"] = page_ids 

58 if space_ids: 

59 params["space-id"] = space_ids 

60 if statuses: 

61 params["status"] = statuses 

62 if title: 

63 params["title"] = title 

64 if sort_condition: 

65 params["sort"] = sort_condition 

66 if limit: 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 params["limit"] = limit 

68 

69 return self._paginate(url, params) 

70 

71 def get_blogposts( 

72 self, 

73 post_ids: List[int] = None, 

74 space_ids: List[str] = None, 

75 statuses: List[str] = None, 

76 title: str = None, 

77 sort_condition: str = None, 

78 limit: int = None, 

79 ) -> List[dict]: 

80 url = f"{self.url}/wiki/api/v2/blogposts" 

81 params = { 

82 "body-format": "storage", 

83 } 

84 if post_ids: 

85 params["id"] = post_ids 

86 if space_ids: 

87 params["space-id"] = space_ids 

88 if statuses: 

89 params["status"] = statuses 

90 if title: 

91 params["title"] = title 

92 if sort_condition: 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true

93 params["sort"] = sort_condition 

94 if limit: 94 ↛ 95line 94 didn't jump to line 95 because the condition on line 94 was never true

95 params["limit"] = limit 

96 

97 return self._paginate(url, params) 

98 

99 def get_whiteboard_by_id(self, whiteboard_id: int) -> dict: 

100 url = f"{self.url}/wiki/api/v2/whiteboards/{whiteboard_id}" 

101 

102 return self._make_request("GET", url) 

103 

104 def get_database_by_id(self, database_id: int) -> dict: 

105 url = f"{self.url}/wiki/api/v2/databases/{database_id}" 

106 

107 return self._make_request("GET", url) 

108 

109 def get_tasks( 

110 self, 

111 task_ids: List[int] = None, 

112 space_ids: List[str] = None, 

113 page_ids: List[str] = None, 

114 blogpost_ids: List[str] = None, 

115 created_by_ids: List[str] = None, 

116 assigned_to_ids: List[str] = None, 

117 completed_by_ids: List[str] = None, 

118 status: str = None, 

119 limit: int = None, 

120 ) -> List[dict]: 

121 url = f"{self.url}/wiki/api/v2/tasks" 

122 params = { 

123 "body-format": "storage", 

124 } 

125 if task_ids: 

126 params["id"] = task_ids 

127 if space_ids: 

128 params["space-id"] = space_ids 

129 if page_ids: 

130 params["page-id"] = page_ids 

131 if blogpost_ids: 131 ↛ 132line 131 didn't jump to line 132 because the condition on line 131 was never true

132 params["blogpost-id"] = blogpost_ids 

133 if created_by_ids: 

134 params["created-by"] = created_by_ids 

135 if assigned_to_ids: 

136 params["assigned-to"] = assigned_to_ids 

137 if completed_by_ids: 

138 params["completed-by"] = completed_by_ids 

139 if status: 

140 params["status"] = status 

141 if limit: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true

142 params["limit"] = limit 

143 

144 return self._paginate(url, params) 

145 

146 def _paginate(self, url: str, params: dict = None) -> List[dict]: 

147 results = [] 

148 response = self._make_request("GET", url, params) 

149 results.extend(response["results"]) 

150 

151 while response["_links"].get("next"): 

152 next_url = response["_links"].get("next") 

153 next_params = {} 

154 if params: 154 ↛ 156line 154 didn't jump to line 156 because the condition on line 154 was always true

155 next_params.update(params) 

156 if "cursor=" in next_url: 

157 # cursor= is 7 characters long 

158 cursor_start = next_url.find("cursor=") + 7 

159 cursor_value = next_url[cursor_start:] 

160 if "&" in cursor_value: 160 ↛ 162line 160 didn't jump to line 162 because the condition on line 160 was always true

161 cursor_value = cursor_value.split("&")[0] 

162 next_params["cursor"] = cursor_value 

163 response = self._make_request("GET", url, next_params) 

164 else: 

165 response = self._make_request("GET", next_url) 

166 results.extend(response["results"]) 

167 

168 return results 

169 

170 def _make_request(self, method: str, url: str, params: dict = None, data: dict = None) -> dict: 

171 response = self.session.request(method, url, params=params, json=data) 

172 

173 if response.status_code != 200: 

174 raise Exception(f"Request failed with status code {response.status_code}: {response.text}") 

175 

176 return response.json()