Coverage for .tox/coverage/lib/python3.11/site-packages/wuttatell/client.py: 100%

1# -*- coding: utf-8; -*- 

2################################################################################ 

3# 

4# WuttaTell -- Telemetry submission for Wutta Framework 

5# Copyright © 2025 Lance Edgar 

6# 

7# This file is part of Wutta Framework. 

8# 

9# Wutta Framework is free software: you can redistribute it and/or modify it 

10# under the terms of the GNU General Public License as published by the Free 

11# Software Foundation, either version 3 of the License, or (at your option) any 

12# later version. 

13# 

14# Wutta Framework is distributed in the hope that it will be useful, but 

15# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 

16# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for 

17# more details. 

18# 

19# You should have received a copy of the GNU General Public License along with 

20# Wutta Framework. If not, see <http://www.gnu.org/licenses/>. 

21# 

22################################################################################ 

23""" 

24Simple API Client 

25""" 

26 

27import json 

28from urllib.parse import urlparse 

29 

30import requests 

31 

32 

33class SimpleAPIClient: 

34 """ 

35 Simple client for "typical" API service. 

36 

37 This basically assumes telemetry can be submitted to a single API 

38 endpoint, and the request should contain an auth token. 

39 

40 :param config: App :term:`config object`. 

41 

42 :param base_url: Base URL of the API. 

43 

44 :param token: Auth token for the API. 

45 

46 :param ssl_verify: Whether the SSL cert presented by the server 

47 should be verified. This is effectively true by default, but 

48 may be disabled for testing with self-signed certs etc. 

49 

50 :param max_retries: Maximum number of retries each connection 

51 should attempt. This value is ultimately given to the 

52 :class:`~requests:requests.adapters.HTTPAdapter` instance. 

53 

54 Most params may be omitted, if config specifies instead: 

55 

56 .. code-block:: ini 

57 

58 [wutta.api] 

59 base_url = https://my.example.com/api 

60 token = XYZPDQ12345 

61 ssl_verify = false 

62 max_retries = 5 

63 

64 Upon instantiation, :attr:`session` will be ``None`` until the 

65 first request is made. (Technically when :meth:`init_session()` 

66 first happens.) 

67 

68 .. attribute:: session 

69 

70 :class:`requests:requests.Session` instance being used to make 

71 API requests. 

72 """ 

73 

74 def __init__( # pylint: disable=too-many-arguments,too-many-positional-arguments 

75 self, config, base_url=None, token=None, ssl_verify=None, max_retries=None 

76 ): 

77 self.config = config 

78 

79 self.base_url = base_url or self.config.require( 

80 f"{self.config.appname}.api.base_url" 

81 ) 

82 self.base_url = self.base_url.rstrip("/") 

83 self.token = token or self.config.require(f"{self.config.appname}.api.token") 

84 

85 if max_retries is not None: 

86 self.max_retries = max_retries 

87 else: 

88 self.max_retries = self.config.get_int( 

89 f"{self.config.appname}.api.max_retries" 

90 ) 

91 

92 if ssl_verify is not None: 

93 self.ssl_verify = ssl_verify 

94 else: 

95 self.ssl_verify = self.config.get_bool( 

96 f"{self.config.appname}.api.ssl_verify", default=True 

97 ) 

98 

99 self.session = None 

100 

101 def init_session(self): 

102 """ 

103 Initialize the HTTP session with the API. 

104 

105 This method is invoked as part of :meth:`make_request()`. 

106 

107 It first checks :attr:`session` and will skip if already initialized. 

108 

109 For initialization, it establishes a new 

110 :class:`requests:requests.Session` instance, and modifies it 

111 as needed per config. 

112 """ 

113 if self.session: 

114 return 

115 

116 self.session = requests.Session() 

117 

118 # maybe *disable* SSL cert verification 

119 # (should only be used for testing e.g. w/ self-signed certs) 

120 if not self.ssl_verify: 

121 self.session.verify = False 

122 

123 # maybe set max retries, e.g. for flaky connections 

124 if self.max_retries is not None: 

125 adapter = requests.adapters.HTTPAdapter(max_retries=self.max_retries) 

126 self.session.mount(self.base_url, adapter) 

127 

128 # TODO: is this a good idea, or hacky security risk..? 

129 # without it, can get error response: 

130 # 400 Client Error: Bad CSRF Origin for url 

131 parts = urlparse(self.base_url) 

132 self.session.headers.update( 

133 { 

134 "Origin": f"{parts.scheme}://{parts.netloc}", 

135 } 

136 ) 

137 

138 # authenticate via token only (for now?) 

139 self.session.headers.update( 

140 { 

141 "Authorization": f"Bearer {self.token}", 

142 } 

143 ) 

144 

145 def make_request(self, request_method, api_method, params=None, data=None): 

146 """ 

147 Make a request to the API, and return the response. 

148 

149 This first calls :meth:`init_session()` to establish the 

150 session if needed. 

151 

152 :param request_method: HTTP request method; for now only 

153 ``'GET'`` and ``'POST'`` are supported. 

154 

155 :param api_method: API method endpoint to use, 

156 e.g. ``'/my/telemetry'`` 

157 

158 :param params: Dict of query string params for the request, if 

159 applicable. 

160 

161 :param data: Payload data for the request, if applicable. 

162 Should be JSON-serializable, e.g. a list or dict. 

163 

164 :rtype: :class:`requests:requests.Response` instance. 

165 """ 

166 self.init_session() 

167 api_method = api_method.lstrip("/") 

168 url = f"{self.base_url}/{api_method}" 

169 if request_method == "GET": 

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

171 elif request_method == "POST": 

172 response = self.session.post(url, params=params, data=json.dumps(data)) 

173 else: 

174 raise NotImplementedError(f"unsupported request method: {request_method}") 

175 response.raise_for_status() 

176 return response 

177 

178 def get(self, api_method, params=None): 

179 """ 

180 Perform a GET request for the given API method, and return the 

181 response. 

182 

183 This calls :meth:`make_request()` for the heavy lifting. 

184 

185 :param api_method: API method endpoint to use, 

186 e.g. ``'/my/telemetry'`` 

187 

188 :param params: Dict of query string params for the request, if 

189 applicable. 

190 

191 :rtype: :class:`requests:requests.Response` instance. 

192 """ 

193 return self.make_request("GET", api_method, params=params) 

194 

195 def post(self, api_method, **kwargs): 

196 """ 

197 Perform a POST request for the given API method, and return 

198 the response. 

199 

200 This calls :meth:`make_request()` for the heavy lifting. 

201 

202 :param api_method: API method endpoint to use, 

203 e.g. ``'/my/telemetry'`` 

204 

205 :rtype: :class:`requests:requests.Response` instance. 

206 """ 

207 return self.make_request("POST", api_method, **kwargs)