Coverage for website/moneybirdsynchronization/administration.py: 38.18%

1"""The MoneyBird API. 

2 

3This code is largely based on moneybird-python by Jan-Jelle Kester, 

4licensed under the MIT license. The source code of moneybird-python 

5can be found on GitHub: https://github.com/jjkester/moneybird-python. 

6""" 

7 

8import functools 

9import logging 

10import time 

11from abc import ABC, abstractmethod 

12from datetime import datetime 

13from functools import reduce 

14from urllib.parse import urljoin 

15 

16from django.conf import settings 

17 

18import requests 

19 

20logger = logging.getLogger(__name__) 

21 

22 

23class Administration(ABC): 

24 """A MoneyBird administration.""" 

25 

26 administration_id = None 

27 

28 def __init__(self, administration_id: int): 

29 self.administration_id = administration_id 

30 

31 @abstractmethod 

32 def get(self, resource_path: str, params: dict | None = None): 

33 """Do a GET on the Moneybird administration.""" 

34 

35 @abstractmethod 

36 def post(self, resource_path: str, data: dict): 

37 """Do a POST request on the Moneybird administration.""" 

38 

39 @abstractmethod 

40 def patch(self, resource_path: str, data: dict): 

41 """Do a PATCH request on the Moneybird administration.""" 

42 

43 @abstractmethod 

44 def delete(self, resource_path: str, data: dict | None = None): 

45 """Do a DELETE request on the Moneybird administration.""" 

46 

47 class InvalidResourcePath(Exception): 

48 """The given resource path is invalid.""" 

49 

50 class Error(Exception): 

51 """An exception that can be thrown while using the administration.""" 

52 

53 def __init__(self, status_code: int, description: str | None = None): 

54 """Create a new administration error.""" 

55 msg = f"API error {status_code}" 

56 if description: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true

57 msg += f": {description}" 

58 

59 self.status_code = status_code 

60 self.description = description 

61 

62 super().__init__(msg) 

63 

64 class Unauthorized(Error): 

65 """The client has insufficient authorization.""" 

66 

67 class NotFound(Error): 

68 """The client requested a resource that could not be found.""" 

69 

70 class InvalidData(Error): 

71 """The client sent invalid data.""" 

72 

73 class Throttled(Error): 

74 """The client sent too many requests.""" 

75 

76 retry_after: int 

77 

78 class ServerError(Error): 

79 """An error happened on the server.""" 

80 

81 @abstractmethod 

82 def _create_session(self) -> requests.Session: 

83 """Create a new session.""" 

84 

85 def _build_url(self, resource_path: str) -> str: 

86 if resource_path.startswith("/"): 

87 raise Administration.InvalidResourcePath( 

88 "The resource path must not start with a slash." 

89 ) 

90 

91 api_base_url = "https://moneybird.com/api/v2/" 

92 url_parts = [ 

93 api_base_url, 

94 f"{self.administration_id}/", 

95 f"{resource_path}.json", 

96 ] 

97 return reduce(urljoin, url_parts) 

98 

99 def _process_response(self, response: requests.Response) -> dict | None: 

100 logger.debug(f"Response {response.status_code}: {response.text}") 

101 

102 if response.next: 

103 logger.debug(f"Received paginated response: {response.next}") 

104 

105 good_codes = {200, 201, 204} 

106 bad_codes = { 

107 400: Administration.InvalidData, 

108 401: Administration.Unauthorized, 

109 403: Administration.Unauthorized, 

110 404: Administration.NotFound, 

111 406: Administration.InvalidData, 

112 422: Administration.InvalidData, 

113 429: Administration.Throttled, 

114 500: Administration.ServerError, 

115 } 

116 

117 code = response.status_code 

118 

119 code_is_known: bool = code in good_codes | bad_codes.keys() 

120 

121 if not code_is_known: 

122 logger.warning(f"Unknown response code {code}") 

123 raise Administration.Error( 

124 code, "API response contained unknown status code" 

125 ) 

126 

127 if code in bad_codes: 

128 error = bad_codes[code] 

129 if error == Administration.Throttled: 

130 e = Administration.Throttled(code, "Throttled") 

131 e.retry_after = response.headers.get("Retry-After") 

132 e.rate_limit_remaining = response.headers.get("RateLimit-Remaining") 

133 e.rate_limit_limit = response.headers.get("RateLimit-Limit") 

134 e.rate_limit_reset = response.headers.get("RateLimit-Reset") 

135 error_description = f"retry after {e.retry_after}" 

136 else: 

137 try: 

138 error_description = response.json()["error"] 

139 except (AttributeError, TypeError, KeyError, ValueError): 

140 error_description = None 

141 

142 e = error(code, error_description) 

143 

144 logger.warning(f"API error {code}: {e}") 

145 

146 raise e 

147 

148 if code == 204: 

149 return {} 

150 

151 if response.text == "200": 

152 return {} 

153 

154 return response.json() 

155 

156 

157def _retry_if_throttled(): 

158 max_retries = 3 

159 

160 def decorator_retry(func): 

161 @functools.wraps(func) 

162 def wrapper(*args, **kwargs): 

163 retries = 0 

164 while retries < max_retries: 

165 try: 

166 return func(*args, **kwargs) 

167 except HttpsAdministration.Throttled as e: 

168 retries += 1 

169 retry_after = datetime.fromtimestamp(float(e.retry_after)) 

170 now = datetime.now() 

171 sleep_seconds = int(retry_after.timestamp() - now.timestamp()) + 1 

172 if retries < max_retries: 

173 logger.info(f"Retrying in {sleep_seconds} seconds...") 

174 time.sleep(sleep_seconds) 

175 else: 

176 logger.warning("Max retries reached. Giving up.") 

177 return None 

178 

179 return wrapper 

180 

181 return decorator_retry 

182 

183 

184class HttpsAdministration(Administration): 

185 """The HTTPS implementation of the MoneyBird Administration interface.""" 

186 

187 def __init__(self, key: str, administration_id: int): 

188 """Create a new MoneyBird administration connection.""" 

189 super().__init__(administration_id) 

190 self.key = key 

191 self.session = self._create_session() 

192 

193 def _create_session(self) -> requests.Session: 

194 session = requests.Session() 

195 session.headers.update({"Authorization": f"Bearer {self.key}"}) 

196 return session 

197 

198 @_retry_if_throttled() 

199 def get(self, resource_path: str, params: dict | None = None): 

200 """Do a GET on the Moneybird administration.""" 

201 url = self._build_url(resource_path) 

202 logger.debug(f"GET {url} {params}") 

203 response = self.session.get(url, params=params) 

204 return self._process_response(response) 

205 

206 @_retry_if_throttled() 

207 def post(self, resource_path: str, data: dict): 

208 """Do a POST request for json data on the Moneybird administration.""" 

209 url = self._build_url(resource_path) 

210 logger.debug(f"POST {url} with {data}") 

211 response = self.session.post(url, json=data) 

212 return self._process_response(response) 

213 

214 @_retry_if_throttled() 

215 def post_files(self, resource_path: str, files: dict[str, bytes]): 

216 """Do a POST request for a file or other data on the Moneybird administration.""" 

217 url = self._build_url(resource_path) 

218 logger.debug(f"POST {url} with {files}") 

219 response = self.session.post(url, files=files) 

220 return self._process_response(response) 

221 

222 @_retry_if_throttled() 

223 def patch(self, resource_path: str, data: dict): 

224 """Do a PATCH request on the Moneybird administration.""" 

225 url = self._build_url(resource_path) 

226 logger.debug(f"PATCH {url} with {data}") 

227 response = self.session.patch(url, json=data) 

228 return self._process_response(response) 

229 

230 @_retry_if_throttled() 

231 def delete(self, resource_path: str, data: dict | None = None): 

232 """Do a DELETE on the Moneybird administration.""" 

233 url = self._build_url(resource_path) 

234 logger.debug(f"DELETE {url}") 

235 response = self.session.delete(url, json=data) 

236 return self._process_response(response) 

237 

238 

239class MoneybirdNotConfiguredError(RuntimeError): 

240 pass 

241 

242 

243def get_moneybird_administration(): 

244 if settings.MONEYBIRD_ADMINISTRATION_ID and settings.MONEYBIRD_API_KEY: 

245 return HttpsAdministration( 

246 settings.MONEYBIRD_API_KEY, settings.MONEYBIRD_ADMINISTRATION_ID 

247 ) 

248 raise MoneybirdNotConfiguredError()