Coverage for fingerprint_pro_server_api_sdk/rest.py: 71%
110 statements
« prev ^ index » next coverage.py v7.6.9, created at 2024-12-09 17:50 +0000
1# coding: utf-8
3"""
4 Fingerprint Pro Server API
6 Fingerprint Pro Server API allows you to get information about visitors and about individual events in a server environment. It can be used for data exports, decision-making, and data analysis scenarios. Server 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. # noqa: E501
8 OpenAPI spec version: 3
9 Contact: support@fingerprint.com
10 Generated by: https://github.com/swagger-api/swagger-codegen.git
11"""
13import io
14import json
15import logging
16import re
17import ssl
18import certifi
20from typing import Optional, Dict, List, Tuple, Any
21from urllib.parse import urlencode
23from fingerprint_pro_server_api_sdk.configuration import Configuration
25try:
26 import urllib3
27except ImportError:
28 raise ImportError('Swagger python client requires urllib3.')
31logger = logging.getLogger(__name__)
34class RESTResponse(io.IOBase):
36 def __init__(self, resp: urllib3.HTTPResponse):
37 self.urllib3_response = resp
38 self.status = resp.status
39 self.reason = resp.reason
40 self.data = resp.data
42 def getheaders(self):
43 """Returns a dictionary of the response headers."""
44 return self.urllib3_response.headers
46 def getheader(self, name, default=None):
47 """Returns a given response header."""
48 return self.urllib3_response.headers.get(name, default)
51class RESTClientObject:
53 def __init__(self, configuration: Configuration, pools_size: int = 4, maxsize: int = None):
54 # urllib3.PoolManager will pass all kw parameters to connectionpool
55 # https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/poolmanager.py#L75 # noqa: E501
56 # https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/connectionpool.py#L680 # noqa: E501
57 # maxsize is the number of requests to host that are allowed in parallel # noqa: E501
58 # Custom SSL certificates and client certificates: http://urllib3.readthedocs.io/en/latest/advanced-usage.html # noqa: E501
60 # cert_reqs
61 if configuration.verify_ssl:
62 cert_reqs = ssl.CERT_REQUIRED
63 else:
64 cert_reqs = ssl.CERT_NONE
66 # ca_certs
67 if configuration.ssl_ca_cert:
68 ca_certs = configuration.ssl_ca_cert
69 else:
70 # if not set certificate file, use Mozilla's root certificates.
71 ca_certs = certifi.where()
73 addition_pool_args = {}
74 if configuration.assert_hostname is not None:
75 addition_pool_args['assert_hostname'] = configuration.assert_hostname # noqa: E501
77 if maxsize is None:
78 if configuration.connection_pool_maxsize is not None:
79 maxsize = configuration.connection_pool_maxsize
80 else:
81 maxsize = 4
83 # https pool manager
84 if configuration.proxy:
85 self.pool_manager = urllib3.ProxyManager(
86 num_pools=pools_size,
87 maxsize=maxsize,
88 cert_reqs=cert_reqs,
89 ca_certs=ca_certs,
90 cert_file=configuration.cert_file,
91 key_file=configuration.key_file,
92 proxy_url=configuration.proxy,
93 **addition_pool_args
94 )
95 else:
96 self.pool_manager = urllib3.PoolManager(
97 num_pools=pools_size,
98 maxsize=maxsize,
99 cert_reqs=cert_reqs,
100 ca_certs=ca_certs,
101 cert_file=configuration.cert_file,
102 key_file=configuration.key_file,
103 **addition_pool_args
104 )
106 def request(self, method: str, url: str, query_params: Optional[List[Tuple[str, Any]]] = None,
107 headers: Optional[Dict[str, Any]] = None, body: Any = None,
108 post_params: Optional[List[Tuple[str, Any]]] = None,
109 _preload_content: bool = True, _request_timeout: Optional[int] = None):
110 """Perform requests.
112 :param method: http request method
113 :param url: http request url
114 :param query_params: query parameters in the url
115 :param headers: http request headers
116 :param body: request json body, for `application/json`
117 :param post_params: request post parameters,
118 `application/x-www-form-urlencoded`
119 and `multipart/form-data`
120 :param _preload_content: if False, the urllib3.HTTPResponse object will
121 be returned without reading/decoding response
122 data. Default is True.
123 :param _request_timeout: timeout setting for this request. If one
124 number provided, it will be total request
125 timeout. It can also be a pair (tuple) of
126 (connection, read) timeouts.
127 """
128 method = method.upper()
129 assert method in ['GET', 'HEAD', 'DELETE', 'POST', 'PUT',
130 'PATCH', 'OPTIONS']
132 if post_params and body:
133 raise ValueError(
134 "body parameter cannot be used with post_params parameter."
135 )
137 post_params = post_params or {}
138 headers = headers or {}
140 timeout = None
141 if _request_timeout:
142 if isinstance(_request_timeout, int):
143 timeout = urllib3.Timeout(total=_request_timeout)
144 elif (isinstance(_request_timeout, tuple) and
145 len(_request_timeout) == 2):
146 timeout = urllib3.Timeout(
147 connect=_request_timeout[0], read=_request_timeout[1])
149 if 'Content-Type' not in headers:
150 headers['Content-Type'] = 'application/json'
152 try:
153 # For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE`
154 if method in ['POST', 'PUT', 'PATCH', 'OPTIONS', 'DELETE']:
155 if query_params:
156 url += '?' + urlencode(query_params)
157 if re.search('json', headers['Content-Type'], re.IGNORECASE):
158 request_body = '{}'
159 if body is not None:
160 request_body = json.dumps(body)
161 r = self.pool_manager.request(
162 method, url,
163 body=request_body,
164 preload_content=_preload_content,
165 timeout=timeout,
166 headers=headers)
167 elif headers['Content-Type'] == 'application/x-www-form-urlencoded': # noqa: E501
168 r = self.pool_manager.request(
169 method, url,
170 fields=post_params,
171 encode_multipart=False,
172 preload_content=_preload_content,
173 timeout=timeout,
174 headers=headers)
175 elif headers['Content-Type'] == 'multipart/form-data':
176 # must del headers['Content-Type'], or the correct
177 # Content-Type which generated by urllib3 will be
178 # overwritten.
179 del headers['Content-Type']
180 r = self.pool_manager.request(
181 method, url,
182 fields=post_params,
183 encode_multipart=True,
184 preload_content=_preload_content,
185 timeout=timeout,
186 headers=headers)
187 # Pass a `string` parameter directly in the body to support
188 # other content types than Json when `body` argument is
189 # provided in serialized form
190 elif isinstance(body, str):
191 request_body = body
192 r = self.pool_manager.request(
193 method, url,
194 body=request_body,
195 preload_content=_preload_content,
196 timeout=timeout,
197 headers=headers)
198 else:
199 # Cannot generate the request from given parameters
200 msg = """Cannot prepare a request message for provided
201 arguments. Please check that your arguments match
202 declared content type."""
203 raise ApiException(status=0, reason=msg)
204 # For `GET`, `HEAD`
205 else:
206 r = self.pool_manager.request(method, url,
207 fields=query_params,
208 preload_content=_preload_content,
209 timeout=timeout,
210 headers=headers)
211 except urllib3.exceptions.SSLError as e:
212 msg = "{0}\n{1}".format(type(e).__name__, str(e))
213 raise ApiException(status=0, reason=msg)
215 if _preload_content:
216 r = RESTResponse(r)
218 # log response body
219 logger.debug("response body: %s", r.data)
221 if not 200 <= r.status <= 299:
222 raise ApiException(http_resp=r)
224 return r
227class ApiException(Exception):
229 def __init__(self, status: Optional[int] = None, reason: Optional[str] = None,
230 http_resp: Optional[RESTResponse] = None):
231 if http_resp:
232 self.status = http_resp.status
233 self.reason = http_resp.reason
234 self.body = http_resp.data
235 self.headers = http_resp.getheaders()
236 else:
237 self.status = status
238 self.reason = reason
239 self.body = None
240 self.headers = None
242 def __str__(self) -> str:
243 """Custom error messages for exception"""
244 error_message = "({0})\n"\
245 "Reason: {1}\n".format(self.status, self.reason)
246 if self.headers:
247 error_message += "HTTP response headers: {0}\n".format(
248 self.headers)
250 if self.body:
251 error_message += "HTTP response body: {0}\n".format(self.body)
253 return error_message
256class KnownApiException(ApiException):
258 def __init__(self, original_exception: ApiException, structured_error: Any):
259 super(KnownApiException, self).__init__(status=original_exception.status, reason=original_exception.reason)
260 self.headers = original_exception.headers
261 self.body = original_exception.body
262 self.structured_error = structured_error