Coverage for fingerprint_server_sdk/rest.py: 60%

1""" 

2Server API 

3Fingerprint Server API allows you to get, search, and update Events in a server environment. It can be used for data exports, decision-making, and data analysis scenarios. 

4Server API is intended for server-side usage, it's not intended to be used from the client side, whether it's a browser or a mobile device. 

5The API also supports collection of Automation Intelligence for requests to your server in edge, pre-origin, or middleware contexts. 

6 

7The version of the OpenAPI document: 4 

8Contact: support@fingerprint.com 

9Generated by OpenAPI Generator (https://openapi-generator.tech) 

10 

11Do not edit the class manually. 

12""" # noqa: E501 

13 

14from __future__ import annotations 

15 

16import io 

17import json 

18import re 

19import ssl 

20from typing import TYPE_CHECKING, Any, Optional, Union 

21 

22import urllib3 

23 

24from fingerprint_server_sdk.exceptions import ApiException, ApiValueError 

25 

26if TYPE_CHECKING: 

27 from fingerprint_server_sdk.configuration import Configuration 

28 

29SUPPORTED_SOCKS_PROXIES = {'socks5', 'socks5h', 'socks4', 'socks4a'} 

30RESTResponseType = urllib3.HTTPResponse 

31 

32 

33def is_socks_proxy_url(url: Optional[str]) -> bool: 

34 if url is None: 

35 return False 

36 split_section = url.split('://') 

37 if len(split_section) < 2: 

38 return False 

39 else: 

40 return split_section[0].lower() in SUPPORTED_SOCKS_PROXIES 

41 

42 

43class RESTResponse(io.IOBase): 

44 def __init__(self, resp: urllib3.HTTPResponse) -> None: 

45 self.response = resp 

46 self.status = resp.status 

47 self.reason = resp.reason 

48 self.data: Optional[bytes] = None 

49 

50 def read(self) -> bytes: 

51 if self.data is None: 

52 self.data = self.response.data 

53 return self.data 

54 

55 @property 

56 def headers(self) -> urllib3.HTTPHeaderDict: 

57 """Returns a dictionary of response headers.""" 

58 return self.response.headers 

59 

60 def getheaders(self) -> urllib3.HTTPHeaderDict: 

61 """Returns a dictionary of the response headers; use ``headers`` instead.""" 

62 return self.response.headers 

63 

64 def getheader(self, name: str, default: Optional[str] = None) -> Optional[str]: 

65 """Returns a given response header; use ``headers.get()`` instead.""" 

66 return self.response.headers.get(name, default) 

67 

68 

69class RESTClientObject: 

70 def __init__(self, configuration: Configuration) -> None: 

71 # urllib3.PoolManager will pass all kw parameters to connectionpool 

72 # https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/poolmanager.py#L75 # noqa: E501 

73 # https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/connectionpool.py#L680 # noqa: E501 

74 # Custom SSL certificates and client certificates: http://urllib3.readthedocs.io/en/latest/advanced-usage.html # noqa: E501 

75 

76 # cert_reqs 

77 if configuration.verify_ssl: 

78 cert_reqs = ssl.CERT_REQUIRED 

79 else: 

80 cert_reqs = ssl.CERT_NONE 

81 

82 pool_args: dict[str, Any] = { 

83 'cert_reqs': cert_reqs, 

84 'ca_certs': configuration.ssl_ca_cert, 

85 'cert_file': configuration.cert_file, 

86 'key_file': configuration.key_file, 

87 'ca_cert_data': configuration.ca_cert_data, 

88 } 

89 if configuration.assert_hostname is not None: 

90 pool_args['assert_hostname'] = configuration.assert_hostname 

91 

92 if configuration.retries is not None: 

93 pool_args['retries'] = configuration.retries 

94 

95 if configuration.tls_server_name: 

96 pool_args['server_hostname'] = configuration.tls_server_name 

97 

98 if configuration.socket_options is not None: 

99 pool_args['socket_options'] = configuration.socket_options 

100 

101 if configuration.connection_pool_maxsize is not None: 

102 pool_args['maxsize'] = configuration.connection_pool_maxsize 

103 

104 # https pool manager 

105 self.pool_manager: urllib3.PoolManager 

106 

107 if configuration.proxy: 

108 if is_socks_proxy_url(configuration.proxy): 

109 from urllib3.contrib.socks import SOCKSProxyManager 

110 

111 pool_args['proxy_url'] = configuration.proxy 

112 pool_args['headers'] = configuration.proxy_headers 

113 self.pool_manager = SOCKSProxyManager(**pool_args) 

114 else: 

115 pool_args['proxy_url'] = configuration.proxy 

116 pool_args['proxy_headers'] = configuration.proxy_headers 

117 self.pool_manager = urllib3.ProxyManager(**pool_args) 

118 else: 

119 self.pool_manager = urllib3.PoolManager(**pool_args) 

120 

121 def request( 

122 self, 

123 method: str, 

124 url: str, 

125 headers: Optional[dict[str, str]] = None, 

126 body: Optional[Any] = None, 

127 post_params: Optional[list[tuple[str, Any]]] = None, 

128 _request_timeout: Optional[Union[float, tuple[float, float]]] = None, 

129 ) -> RESTResponse: 

130 """Perform requests. 

131 

132 :param method: http request method 

133 :param url: http request url 

134 :param headers: http request headers 

135 :param body: request json body, for `application/json` 

136 :param post_params: request post parameters, 

137 `application/x-www-form-urlencoded` 

138 and `multipart/form-data` 

139 :param _request_timeout: timeout setting for this request. If one 

140 number provided, it will be total request 

141 timeout. It can also be a pair (tuple) of 

142 (connection, read) timeouts. 

143 """ 

144 method = method.upper() 

145 assert method in ['GET', 'HEAD', 'DELETE', 'POST', 'PUT', 'PATCH', 'OPTIONS'] 

146 

147 if post_params and body: 

148 raise ApiValueError('body parameter cannot be used with post_params parameter.') 

149 

150 post_params = post_params or [] 

151 headers = headers or {} 

152 

153 timeout = None 

154 if _request_timeout: 

155 if isinstance(_request_timeout, (int, float)): 

156 timeout = urllib3.Timeout(total=_request_timeout) 

157 elif isinstance(_request_timeout, tuple) and len(_request_timeout) == 2: 

158 timeout = urllib3.Timeout(connect=_request_timeout[0], read=_request_timeout[1]) 

159 

160 try: 

161 # For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE` 

162 if method in ['POST', 'PUT', 'PATCH', 'OPTIONS', 'DELETE']: 

163 # no content type provided or payload is json 

164 content_type = headers.get('Content-Type') 

165 if not content_type or re.search('json', content_type, re.IGNORECASE): 

166 request_body = None 

167 if body is not None: 

168 request_body = json.dumps(body) 

169 r = self.pool_manager.request( 

170 method, 

171 url, 

172 body=request_body, 

173 timeout=timeout, 

174 headers=headers, 

175 preload_content=False, 

176 ) 

177 elif content_type == 'application/x-www-form-urlencoded': 

178 r = self.pool_manager.request( 

179 method, 

180 url, 

181 fields=post_params, 

182 encode_multipart=False, 

183 timeout=timeout, 

184 headers=headers, 

185 preload_content=False, 

186 ) 

187 elif content_type == 'multipart/form-data': 

188 # must del headers['Content-Type'], or the correct 

189 # Content-Type which generated by urllib3 will be 

190 # overwritten. 

191 del headers['Content-Type'] 

192 # Ensures that dict objects are serialized 

193 post_params = [ 

194 (a, json.dumps(b)) if isinstance(b, dict) else (a, b) 

195 for a, b in post_params 

196 ] 

197 r = self.pool_manager.request( 

198 method, 

199 url, 

200 fields=post_params, 

201 encode_multipart=True, 

202 timeout=timeout, 

203 headers=headers, 

204 preload_content=False, 

205 ) 

206 # Pass a `string` parameter directly in the body to support 

207 # other content types than JSON when `body` argument is 

208 # provided in serialized form. 

209 elif isinstance(body, (str, bytes)): 

210 r = self.pool_manager.request( 

211 method, 

212 url, 

213 body=body, 

214 timeout=timeout, 

215 headers=headers, 

216 preload_content=False, 

217 ) 

218 elif headers['Content-Type'].startswith('text/') and isinstance(body, bool): 

219 request_body = 'true' if body else 'false' 

220 r = self.pool_manager.request( 

221 method, 

222 url, 

223 body=request_body, 

224 preload_content=False, 

225 timeout=timeout, 

226 headers=headers, 

227 ) 

228 else: 

229 # Cannot generate the request from given parameters 

230 msg = """Cannot prepare a request message for provided 

231 arguments. Please check that your arguments match 

232 declared content type.""" 

233 raise ApiException(status=0, reason=msg) 

234 # For `GET`, `HEAD` 

235 else: 

236 r = self.pool_manager.request( 

237 method, url, fields={}, timeout=timeout, headers=headers, preload_content=False 

238 ) 

239 except urllib3.exceptions.SSLError as e: 

240 msg = '\n'.join([type(e).__name__, str(e)]) 

241 raise ApiException(status=0, reason=msg) from e 

242 

243 return RESTResponse(r) # type: ignore[arg-type]