Coverage for fingerprint_server_sdk / rest.py: 60%
100 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-11 18:41 +0000
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.
6The version of the OpenAPI document: 4
7Contact: support@fingerprint.com
8Generated by OpenAPI Generator (https://openapi-generator.tech)
10Do not edit the class manually.
11""" # noqa: E501
13from __future__ import annotations
15import io
16import json
17import re
18import ssl
19from typing import TYPE_CHECKING, Any, Optional, Union
21import urllib3
23from fingerprint_server_sdk.exceptions import ApiException, ApiValueError
25if TYPE_CHECKING:
26 from fingerprint_server_sdk.configuration import Configuration
28SUPPORTED_SOCKS_PROXIES = {'socks5', 'socks5h', 'socks4', 'socks4a'}
29RESTResponseType = urllib3.HTTPResponse
32def is_socks_proxy_url(url: Optional[str]) -> bool:
33 if url is None:
34 return False
35 split_section = url.split('://')
36 if len(split_section) < 2:
37 return False
38 else:
39 return split_section[0].lower() in SUPPORTED_SOCKS_PROXIES
42class RESTResponse(io.IOBase):
43 def __init__(self, resp: urllib3.HTTPResponse) -> None:
44 self.response = resp
45 self.status = resp.status
46 self.reason = resp.reason
47 self.data: Optional[bytes] = None
49 def read(self) -> bytes:
50 if self.data is None:
51 self.data = self.response.data
52 return self.data
54 @property
55 def headers(self) -> urllib3.HTTPHeaderDict:
56 """Returns a dictionary of response headers."""
57 return self.response.headers
59 def getheaders(self) -> urllib3.HTTPHeaderDict:
60 """Returns a dictionary of the response headers; use ``headers`` instead."""
61 return self.response.headers
63 def getheader(self, name: str, default: Optional[str] = None) -> Optional[str]:
64 """Returns a given response header; use ``headers.get()`` instead."""
65 return self.response.headers.get(name, default)
68class RESTClientObject:
69 def __init__(self, configuration: Configuration) -> None:
70 # urllib3.PoolManager will pass all kw parameters to connectionpool
71 # https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/poolmanager.py#L75 # noqa: E501
72 # https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/connectionpool.py#L680 # noqa: E501
73 # Custom SSL certificates and client certificates: http://urllib3.readthedocs.io/en/latest/advanced-usage.html # noqa: E501
75 # cert_reqs
76 if configuration.verify_ssl:
77 cert_reqs = ssl.CERT_REQUIRED
78 else:
79 cert_reqs = ssl.CERT_NONE
81 pool_args: dict[str, Any] = {
82 'cert_reqs': cert_reqs,
83 'ca_certs': configuration.ssl_ca_cert,
84 'cert_file': configuration.cert_file,
85 'key_file': configuration.key_file,
86 'ca_cert_data': configuration.ca_cert_data,
87 }
88 if configuration.assert_hostname is not None:
89 pool_args['assert_hostname'] = configuration.assert_hostname
91 if configuration.retries is not None:
92 pool_args['retries'] = configuration.retries
94 if configuration.tls_server_name:
95 pool_args['server_hostname'] = configuration.tls_server_name
97 if configuration.socket_options is not None:
98 pool_args['socket_options'] = configuration.socket_options
100 if configuration.connection_pool_maxsize is not None:
101 pool_args['maxsize'] = configuration.connection_pool_maxsize
103 # https pool manager
104 self.pool_manager: urllib3.PoolManager
106 if configuration.proxy:
107 if is_socks_proxy_url(configuration.proxy):
108 from urllib3.contrib.socks import SOCKSProxyManager
110 pool_args['proxy_url'] = configuration.proxy
111 pool_args['headers'] = configuration.proxy_headers
112 self.pool_manager = SOCKSProxyManager(**pool_args)
113 else:
114 pool_args['proxy_url'] = configuration.proxy
115 pool_args['proxy_headers'] = configuration.proxy_headers
116 self.pool_manager = urllib3.ProxyManager(**pool_args)
117 else:
118 self.pool_manager = urllib3.PoolManager(**pool_args)
120 def request(
121 self,
122 method: str,
123 url: str,
124 headers: Optional[dict[str, str]] = None,
125 body: Optional[Any] = None,
126 post_params: Optional[list[tuple[str, Any]]] = None,
127 _request_timeout: Optional[Union[float, tuple[float, float]]] = None,
128 ) -> RESTResponse:
129 """Perform requests.
131 :param method: http request method
132 :param url: http request url
133 :param headers: http request headers
134 :param body: request json body, for `application/json`
135 :param post_params: request post parameters,
136 `application/x-www-form-urlencoded`
137 and `multipart/form-data`
138 :param _request_timeout: timeout setting for this request. If one
139 number provided, it will be total request
140 timeout. It can also be a pair (tuple) of
141 (connection, read) timeouts.
142 """
143 method = method.upper()
144 assert method in ['GET', 'HEAD', 'DELETE', 'POST', 'PUT', 'PATCH', 'OPTIONS']
146 if post_params and body:
147 raise ApiValueError('body parameter cannot be used with post_params parameter.')
149 post_params = post_params or []
150 headers = headers or {}
152 timeout = None
153 if _request_timeout:
154 if isinstance(_request_timeout, (int, float)):
155 timeout = urllib3.Timeout(total=_request_timeout)
156 elif isinstance(_request_timeout, tuple) and len(_request_timeout) == 2:
157 timeout = urllib3.Timeout(connect=_request_timeout[0], read=_request_timeout[1])
159 try:
160 # For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE`
161 if method in ['POST', 'PUT', 'PATCH', 'OPTIONS', 'DELETE']:
162 # no content type provided or payload is json
163 content_type = headers.get('Content-Type')
164 if not content_type or re.search('json', content_type, re.IGNORECASE):
165 request_body = None
166 if body is not None:
167 request_body = json.dumps(body)
168 r = self.pool_manager.request(
169 method,
170 url,
171 body=request_body,
172 timeout=timeout,
173 headers=headers,
174 preload_content=False,
175 )
176 elif content_type == 'application/x-www-form-urlencoded':
177 r = self.pool_manager.request(
178 method,
179 url,
180 fields=post_params,
181 encode_multipart=False,
182 timeout=timeout,
183 headers=headers,
184 preload_content=False,
185 )
186 elif content_type == 'multipart/form-data':
187 # must del headers['Content-Type'], or the correct
188 # Content-Type which generated by urllib3 will be
189 # overwritten.
190 del headers['Content-Type']
191 # Ensures that dict objects are serialized
192 post_params = [
193 (a, json.dumps(b)) if isinstance(b, dict) else (a, b)
194 for a, b in post_params
195 ]
196 r = self.pool_manager.request(
197 method,
198 url,
199 fields=post_params,
200 encode_multipart=True,
201 timeout=timeout,
202 headers=headers,
203 preload_content=False,
204 )
205 # Pass a `string` parameter directly in the body to support
206 # other content types than JSON when `body` argument is
207 # provided in serialized form.
208 elif isinstance(body, (str, bytes)):
209 r = self.pool_manager.request(
210 method,
211 url,
212 body=body,
213 timeout=timeout,
214 headers=headers,
215 preload_content=False,
216 )
217 elif headers['Content-Type'].startswith('text/') and isinstance(body, bool):
218 request_body = 'true' if body else 'false'
219 r = self.pool_manager.request(
220 method,
221 url,
222 body=request_body,
223 preload_content=False,
224 timeout=timeout,
225 headers=headers,
226 )
227 else:
228 # Cannot generate the request from given parameters
229 msg = """Cannot prepare a request message for provided
230 arguments. Please check that your arguments match
231 declared content type."""
232 raise ApiException(status=0, reason=msg)
233 # For `GET`, `HEAD`
234 else:
235 r = self.pool_manager.request(
236 method, url, fields={}, timeout=timeout, headers=headers, preload_content=False
237 )
238 except urllib3.exceptions.SSLError as e:
239 msg = '\n'.join([type(e).__name__, str(e)])
240 raise ApiException(status=0, reason=msg) from e
242 return RESTResponse(r) # type: ignore[arg-type]