Coverage for fingerprint_server_sdk/api_client.py: 58%
327 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-14 10:45 +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.
5The API also supports collection of Automation Intelligence for requests to your server in edge, pre-origin, or middleware contexts.
7The version of the OpenAPI document: 4
8Contact: support@fingerprint.com
9Generated by OpenAPI Generator (https://openapi-generator.tech)
11Do not edit the class manually.
12""" # noqa: E501
14from __future__ import annotations
16import datetime
17import decimal
18import json
19import mimetypes
20import os
21import re
22import tempfile
23import uuid
24from enum import Enum
25from types import TracebackType
26from typing import Any, Optional, Union
27from urllib.parse import quote
29from dateutil.parser import parse
30from pydantic import SecretStr
32import fingerprint_server_sdk.models
33from fingerprint_server_sdk import __version__, rest
34from fingerprint_server_sdk.api_response import ApiResponse
35from fingerprint_server_sdk.api_response import T as ApiResponseT
36from fingerprint_server_sdk.configuration import Configuration
37from fingerprint_server_sdk.exceptions import (
38 ApiException,
39 ApiValueError,
40)
42RequestSerialized = tuple[str, str, dict[str, Any], Optional[Any], Any]
43FilesType = dict[
44 str, Union[str, bytes, list[str], list[bytes], tuple[str, bytes], list[tuple[str, bytes]]]
45]
48class ApiClient:
49 """Generic API client for OpenAPI client library builds.
51 OpenAPI generic API client. This client handles the client-
52 server communication, and is invariant across implementations. Specifics of
53 the methods and models for each application are generated from the OpenAPI
54 templates.
56 :param configuration: .Configuration object for this client
57 :param header_name: a header to pass when making calls to the API.
58 :param header_value: a header value to pass when making calls to
59 the API.
60 :param cookie: a cookie to include in the header when making calls
61 to the API
62 """
64 PRIMITIVE_TYPES = (float, bool, bytes, str, int)
65 NATIVE_TYPES_MAPPING = {
66 'int': int,
67 'float': float,
68 'str': str,
69 'bool': bool,
70 'date': datetime.date,
71 'datetime': datetime.datetime,
72 'decimal': decimal.Decimal,
73 'object': object,
74 }
76 def __init__(
77 self,
78 configuration: Configuration,
79 header_name: Optional[str] = None,
80 header_value: Optional[str] = None,
81 cookie: Optional[str] = None,
82 ) -> None:
83 self.configuration = configuration
85 self.rest_client = rest.RESTClientObject(configuration)
86 self.default_headers: dict[str, str] = {}
87 if header_name is not None and header_value is not None:
88 self.default_headers[header_name] = header_value
89 self.cookie = cookie
90 # Set default User-Agent.
91 self.user_agent = f'fingerprint-pro-server-python-sdk/{__version__}'
92 self.client_side_validation = configuration.client_side_validation
94 def __enter__(self) -> ApiClient:
95 return self
97 def __exit__(
98 self,
99 exc_type: Optional[type[BaseException]],
100 exc_value: Optional[BaseException],
101 traceback: Optional[TracebackType],
102 ) -> None:
103 pass
105 @property
106 def user_agent(self) -> str:
107 """User agent for this API client"""
108 return self.default_headers['User-Agent']
110 @user_agent.setter
111 def user_agent(self, value: str) -> None:
112 self.default_headers['User-Agent'] = value
114 def set_default_header(self, header_name: str, header_value: str) -> None:
115 self.default_headers[header_name] = header_value
117 def param_serialize(
118 self,
119 method: str,
120 resource_path: str,
121 path_params: Optional[dict[str, Any]] = None,
122 query_params: Optional[list[tuple[str, Any]]] = None,
123 header_params: Optional[dict[str, Any]] = None,
124 body: Optional[Any] = None,
125 post_params: Optional[list[tuple[str, Any]]] = None,
126 files: Optional[FilesType] = None,
127 auth_settings: Optional[list[str]] = None,
128 collection_formats: Optional[dict[str, str]] = None,
129 _request_auth: Optional[dict[str, Any]] = None,
130 ) -> RequestSerialized:
131 """Builds the HTTP request params needed by the request.
132 :param method: Method to call.
133 :param resource_path: Path to method endpoint.
134 :param path_params: Path parameters in the url.
135 :param query_params: Query parameters in the url.
136 :param header_params: Header parameters to be
137 placed in the request header.
138 :param body: Request body.
139 :param post_params dict: Request post form parameters,
140 for `application/x-www-form-urlencoded`, `multipart/form-data`.
141 :param auth_settings list: Auth Settings names for the request.
142 :param files dict: key -> filename, value -> filepath,
143 for `multipart/form-data`.
144 :param collection_formats: dict of collection formats for path, query,
145 header, and post parameters.
146 :param _request_auth: set to override the auth_settings for an a single
147 request; this effectively ignores the authentication
148 in the spec for a single request.
149 :return: tuple of form (path, http_method, query_params, header_params,
150 body, post_params, files)
151 """
153 config = self.configuration
155 # header parameters
156 header_params = header_params or {}
157 header_params.update(self.default_headers)
158 if self.cookie:
159 header_params['Cookie'] = self.cookie
160 if header_params:
161 header_params = self.sanitize_for_serialization(header_params)
162 header_params = dict(self.parameters_to_tuples(header_params, collection_formats))
164 # path parameters
165 if path_params:
166 path_params_sanitized = self.sanitize_for_serialization(path_params)
167 path_params_tuples = self.parameters_to_tuples(
168 path_params_sanitized, collection_formats
169 )
170 for k, v in path_params_tuples:
171 # specified safe chars, encode everything
172 resource_path = resource_path.replace(
173 '{' + k + '}', quote(str(v), safe=config.safe_chars_for_path_param)
174 )
176 # post parameters
177 post_params_result: Any = None
178 if post_params or files:
179 post_params_list = post_params if post_params else []
180 post_params_sanitized = self.sanitize_for_serialization(post_params_list)
181 post_params_result = self.parameters_to_tuples(
182 post_params_sanitized, collection_formats
183 )
184 if files:
185 post_params_result.extend(self.files_parameters(files))
187 # auth setting
188 self.update_params_for_auth(
189 header_params,
190 query_params,
191 auth_settings,
192 resource_path,
193 method,
194 body,
195 request_auth=_request_auth,
196 )
198 # body
199 if body:
200 body = self.sanitize_for_serialization(body)
202 # request url
203 url = self.configuration.host + resource_path
205 query_params = list(query_params or [])
206 if getattr(self.configuration, 'default_query_params', None):
207 existing_keys = {k for k, _ in query_params}
208 for k, v in self.configuration.default_query_params:
209 if k not in existing_keys:
210 query_params.append((k, v))
212 # query parameters
213 if query_params:
214 query_params = self.sanitize_for_serialization(query_params)
215 url_query = self.parameters_to_url_query(query_params, collection_formats)
216 url += '?' + url_query
218 return method, url, header_params, body, post_params_result
220 def call_api(
221 self,
222 method: str,
223 url: str,
224 header_params: Optional[dict[str, Any]] = None,
225 body: Any = None,
226 post_params: Optional[list[Any]] = None,
227 _request_timeout: Optional[Union[int, float, tuple[float, float]]] = None,
228 ) -> rest.RESTResponse:
229 """Makes the HTTP request (synchronous)
230 :param method: Method to call.
231 :param url: Path to method endpoint.
232 :param header_params: Header parameters to be
233 placed in the request header.
234 :param body: Request body.
235 :param post_params: Request post form parameters,
236 for `application/x-www-form-urlencoded`, `multipart/form-data`.
237 :param _request_timeout: timeout setting for this request.
238 :return: RESTResponse
239 """
241 try:
242 # perform request and return response
243 response_data = self.rest_client.request(
244 method,
245 url,
246 headers=header_params,
247 body=body,
248 post_params=post_params,
249 _request_timeout=_request_timeout,
250 )
252 except ApiException as e:
253 raise e
255 return response_data
257 def response_deserialize(
258 self,
259 response_data: rest.RESTResponse,
260 response_types_map: Optional[dict[str, Any]] = None,
261 ) -> ApiResponse[ApiResponseT]:
262 """Deserializes response into an object.
263 :param response_data: RESTResponse object to be deserialized.
264 :param response_types_map: dict of response types.
265 :return: ApiResponse
266 """
268 msg = 'RESTResponse.read() must be called before passing it to response_deserialize()'
269 assert response_data.data is not None, msg
271 if response_types_map is None:
272 response_types_map = {}
274 response_type = response_types_map.get(str(response_data.status))
275 if (
276 not response_type
277 and isinstance(response_data.status, int)
278 and 100 <= response_data.status <= 599
279 ):
280 # if not found, look for '1XX', '2XX', etc.
281 response_type = response_types_map.get(str(response_data.status)[0] + 'XX')
283 # deserialize response data
284 response_text: Optional[str] = None
285 return_data: Any = None
286 try:
287 if response_type == 'bytearray':
288 return_data = response_data.data
289 elif response_type == 'file':
290 return_data = self.__deserialize_file(response_data)
291 elif response_type is not None:
292 match = None
293 content_type = response_data.getheader('content-type')
294 if content_type is not None:
295 match = re.search(r'charset=([a-zA-Z\-\d]+)[\s;]?', content_type)
296 encoding = match.group(1) if match else 'utf-8'
297 response_text = response_data.data.decode(encoding)
298 return_data = self.deserialize(response_text, response_type, content_type)
299 finally:
300 if not 200 <= response_data.status <= 299:
301 raise ApiException.from_response(
302 http_resp=response_data,
303 body=response_text,
304 data=return_data,
305 )
307 return ApiResponse(
308 status_code=response_data.status,
309 data=return_data,
310 headers=response_data.getheaders(),
311 raw_data=response_data.data,
312 )
314 def sanitize_for_serialization(self, obj: Any) -> Any:
315 """Builds a JSON POST object.
317 If obj is None, return None.
318 If obj is SecretStr, return obj.get_secret_value()
319 If obj is str, int, long, float, bool, return directly.
320 If obj is datetime.datetime, datetime.date
321 convert to string in iso8601 format.
322 If obj is decimal.Decimal return string representation.
323 If obj is list, sanitize each element in the list.
324 If obj is dict, return the dict.
325 If obj is OpenAPI model, return the properties dict.
327 :param obj: The data to serialize.
328 :return: The serialized form of data.
329 """
330 if obj is None:
331 return None
332 elif isinstance(obj, Enum):
333 return obj.value
334 elif isinstance(obj, SecretStr):
335 return obj.get_secret_value()
336 elif isinstance(obj, self.PRIMITIVE_TYPES):
337 return obj
338 elif isinstance(obj, uuid.UUID):
339 return str(obj)
340 elif isinstance(obj, list):
341 return [self.sanitize_for_serialization(sub_obj) for sub_obj in obj]
342 elif isinstance(obj, tuple):
343 return tuple(self.sanitize_for_serialization(sub_obj) for sub_obj in obj)
344 elif isinstance(obj, (datetime.datetime, datetime.date)):
345 return obj.isoformat()
346 elif isinstance(obj, decimal.Decimal):
347 return str(obj)
349 elif isinstance(obj, dict):
350 obj_dict = obj
351 else:
352 # Convert model obj to dict except
353 # attributes `openapi_types`, `attribute_map`
354 # and attributes which value is not None.
355 # Convert attribute name to json key in
356 # model definition for request.
357 if hasattr(obj, 'to_dict') and callable(obj.to_dict):
358 obj_dict = obj.to_dict()
359 else:
360 obj_dict = obj.__dict__
362 if isinstance(obj_dict, list):
363 # here we handle instances that can either be a list or something else,
364 # and only became a real list by calling to_dict()
365 return self.sanitize_for_serialization(obj_dict)
367 return {key: self.sanitize_for_serialization(val) for key, val in obj_dict.items()}
369 def deserialize(
370 self, response_text: str, response_type: str, content_type: Optional[str]
371 ) -> Any:
372 """Deserializes response into an object.
374 :param response: RESTResponse object to be deserialized.
375 :param response_type: class literal for
376 deserialized object, or string of class name.
377 :param content_type: content type of response.
379 :return: deserialized object.
380 """
382 # fetch data from response object
383 if content_type is None:
384 try:
385 data = json.loads(response_text)
386 except ValueError:
387 data = response_text
388 elif re.match(
389 r'^application/(json|[\w!#$&.+\-^_]+\+json)\s*(;|$)', content_type, re.IGNORECASE
390 ):
391 if response_text == '':
392 data = ''
393 else:
394 data = json.loads(response_text)
395 elif re.match(r'^text\/[a-z.+-]+\s*(;|$)', content_type, re.IGNORECASE):
396 data = response_text
397 else:
398 raise ApiException(status=0, reason=f'Unsupported content type: {content_type}')
400 return self.__deserialize(data, response_type)
402 def __deserialize(self, data: Any, klass: Any) -> Any:
403 """Deserializes dict, list, str into an object.
405 :param data: dict, list or str.
406 :param klass: class literal, or string of class name.
408 :return: object.
409 """
410 if data is None:
411 return None
413 if isinstance(klass, str):
414 if klass.startswith('List['):
415 m = re.match(r'List\[(.*)]', klass)
416 assert m is not None, 'Malformed List type definition'
417 sub_kls = m.group(1)
418 return [self.__deserialize(sub_data, sub_kls) for sub_data in data]
420 if klass.startswith('Dict['):
421 m = re.match(r'Dict\[([^,]*), (.*)]', klass)
422 assert m is not None, 'Malformed Dict type definition'
423 sub_kls = m.group(2)
424 return {k: self.__deserialize(v, sub_kls) for k, v in data.items()}
426 # convert str to class
427 if klass in self.NATIVE_TYPES_MAPPING:
428 klass = self.NATIVE_TYPES_MAPPING[klass]
429 else:
430 klass = getattr(fingerprint_server_sdk.models, klass)
432 if klass in self.PRIMITIVE_TYPES:
433 return self.__deserialize_primitive(data, klass)
434 elif klass is object:
435 return self.__deserialize_object(data)
436 elif klass is datetime.date:
437 return self.__deserialize_date(data)
438 elif klass is datetime.datetime:
439 return self.__deserialize_datetime(data)
440 elif klass is decimal.Decimal:
441 return decimal.Decimal(data)
442 elif issubclass(klass, Enum):
443 return self.__deserialize_enum(data, klass)
444 else:
445 return self.__deserialize_model(data, klass)
447 def parameters_to_tuples(
448 self,
449 params: Union[dict[str, Any], list[tuple[str, Any]]],
450 collection_formats: Optional[dict[str, str]],
451 ) -> list[tuple[str, str]]:
452 """Get parameters as list of tuples, formatting collections.
454 :param params: Parameters as dict or list of two-tuples
455 :param dict collection_formats: Parameter collection formats
456 :return: Parameters as list of tuples, collections formatted
457 """
458 new_params: list[tuple[str, str]] = []
459 if collection_formats is None:
460 collection_formats = {}
461 for k, v in params.items() if isinstance(params, dict) else params:
462 if k in collection_formats:
463 collection_format = collection_formats[k]
464 if collection_format == 'multi':
465 new_params.extend((k, value) for value in v)
466 else:
467 if collection_format == 'ssv':
468 delimiter = ' '
469 elif collection_format == 'tsv':
470 delimiter = '\t'
471 elif collection_format == 'pipes':
472 delimiter = '|'
473 else: # csv is the default
474 delimiter = ','
475 new_params.append((k, delimiter.join(str(value) for value in v)))
476 else:
477 new_params.append((k, v))
478 return new_params
480 def parameters_to_url_query(
481 self,
482 params: Union[dict[str, Any], list[tuple[str, Any]]],
483 collection_formats: Optional[dict[str, str]],
484 ) -> str:
485 """Get parameters as list of tuples, formatting collections.
487 :param params: Parameters as dict or list of two-tuples
488 :param dict collection_formats: Parameter collection formats
489 :return: URL query string (e.g. a=Hello%20World&b=123)
490 """
491 new_params: list[tuple[str, str]] = []
492 if collection_formats is None:
493 collection_formats = {}
494 for k, v in params.items() if isinstance(params, dict) else params:
495 if isinstance(v, bool):
496 v = str(v).lower()
497 if isinstance(v, (int, float)):
498 v = str(v)
499 if isinstance(v, dict):
500 v = json.dumps(v)
502 if k in collection_formats:
503 collection_format = collection_formats[k]
504 if collection_format == 'multi':
505 new_params.extend((k, quote(str(value))) for value in v)
506 else:
507 if collection_format == 'ssv':
508 delimiter = ' '
509 elif collection_format == 'tsv':
510 delimiter = '\t'
511 elif collection_format == 'pipes':
512 delimiter = '|'
513 else: # csv is the default
514 delimiter = ','
515 new_params.append((k, delimiter.join(quote(str(value)) for value in v)))
516 else:
517 new_params.append((k, quote(str(v))))
519 return '&'.join(['='.join(map(str, item)) for item in new_params])
521 def files_parameters(
522 self,
523 files: FilesType,
524 ) -> list[tuple[Any, Any]]:
525 """Builds form parameters.
527 :param files: File parameters.
528 :return: Form parameters with files.
529 """
530 params = []
531 for k, v in files.items():
532 if isinstance(v, str):
533 with open(v, 'rb') as f:
534 filename = os.path.basename(f.name)
535 filedata = f.read()
536 elif isinstance(v, bytes):
537 filename = k
538 filedata = v
539 elif isinstance(v, tuple):
540 filename, filedata = v
541 elif isinstance(v, list):
542 for file_param in v:
543 params.extend(self.files_parameters({k: file_param}))
544 continue
545 else:
546 raise ValueError('Unsupported file value')
547 mimetype = mimetypes.guess_type(filename)[0] or 'application/octet-stream'
548 params.append((k, (filename, filedata, mimetype)))
549 return params
551 def select_header_accept(self, accepts: list[str]) -> Optional[str]:
552 """Returns `Accept` based on an array of accepts provided.
554 :param accepts: List of headers.
555 :return: Accept (e.g. application/json).
556 """
557 if not accepts:
558 return None
560 for accept in accepts:
561 if re.search('json', accept, re.IGNORECASE):
562 return accept
564 return accepts[0]
566 def select_header_content_type(self, content_types: list[str]) -> Optional[str]:
567 """Returns `Content-Type` based on an array of content_types provided.
569 :param content_types: List of content-types.
570 :return: Content-Type (e.g. application/json).
571 """
572 if not content_types:
573 return None
575 for content_type in content_types:
576 if re.search('json', content_type, re.IGNORECASE):
577 return content_type
579 return content_types[0]
581 def update_params_for_auth(
582 self,
583 headers: dict[str, str],
584 queries: Optional[list[tuple[str, Any]]],
585 auth_settings: Optional[list[str]],
586 resource_path: str,
587 method: str,
588 body: Any,
589 request_auth: Optional[dict[str, Any]] = None,
590 ) -> None:
591 """Updates header and query params based on authentication setting.
593 :param headers: Header parameters dict to be updated.
594 :param queries: Query parameters tuple list to be updated.
595 :param auth_settings: Authentication setting identifiers list.
596 :resource_path: A string representation of the HTTP request resource path.
597 :method: A string representation of the HTTP request method.
598 :body: A object representing the body of the HTTP request.
599 The object type is the return value of sanitize_for_serialization().
600 :param request_auth: if set, the provided settings will
601 override the token in the configuration.
602 """
603 if not auth_settings:
604 return
606 if request_auth:
607 self._apply_auth_params(headers, queries, resource_path, method, body, request_auth)
608 else:
609 for auth in auth_settings:
610 auth_setting = self.configuration.auth_settings().get(auth)
611 if auth_setting:
612 self._apply_auth_params(
613 headers,
614 queries,
615 resource_path,
616 method,
617 body,
618 auth_setting, # type: ignore[arg-type]
619 )
621 def _apply_auth_params(
622 self,
623 headers: dict[str, str],
624 queries: Optional[list[tuple[str, Any]]],
625 resource_path: str,
626 method: str,
627 body: Any,
628 auth_setting: dict[str, Any],
629 ) -> None:
630 """Updates the request parameters based on a single auth_setting
632 :param headers: Header parameters dict to be updated.
633 :param queries: Query parameters tuple list to be updated.
634 :resource_path: A string representation of the HTTP request resource path.
635 :method: A string representation of the HTTP request method.
636 :body: A object representing the body of the HTTP request.
637 The object type is the return value of sanitize_for_serialization().
638 :param auth_setting: auth settings for the endpoint
639 """
640 if auth_setting['in'] == 'cookie':
641 headers['Cookie'] = auth_setting['value']
642 elif auth_setting['in'] == 'header':
643 if auth_setting['type'] != 'http-signature':
644 headers[auth_setting['key']] = auth_setting['value']
645 elif auth_setting['in'] == 'query':
646 if queries is not None:
647 queries.append((auth_setting['key'], auth_setting['value']))
648 else:
649 raise ApiValueError('Authentication token must be in `query` or `header`')
651 def __deserialize_file(self, response: rest.RESTResponse) -> Any:
652 """Deserializes body to file
654 Saves response body into a file in a temporary folder,
655 using the filename from the `Content-Disposition` header if provided.
657 handle file downloading
658 save response body into a tmp file and return the instance
660 :param response: RESTResponse.
661 :return: file path.
662 """
663 fd, path = tempfile.mkstemp(dir=getattr(self.configuration, 'temp_folder_path', None))
664 os.close(fd)
665 os.remove(path)
667 content_disposition = response.getheader('Content-Disposition')
668 if content_disposition:
669 m = re.search(r'filename=[\'"]?([^\'"\s]+)[\'"]?', content_disposition)
670 assert m is not None, "Unexpected 'content-disposition' header value"
671 filename = m.group(1)
672 path = os.path.join(os.path.dirname(path), filename)
674 with open(path, 'wb') as f:
675 if response.data is not None:
676 f.write(response.data)
678 return path
680 def __deserialize_primitive(self, data: Any, klass: type) -> Any:
681 """Deserializes string to primitive type.
683 :param data: str.
684 :param klass: class literal.
686 :return: int, long, float, str, bool.
687 """
688 try:
689 return klass(data)
690 except UnicodeEncodeError:
691 return str(data)
692 except TypeError:
693 return data
695 def __deserialize_object(self, value: Any) -> Any:
696 """Return an original value.
698 :return: object.
699 """
700 return value
702 def __deserialize_date(self, string: str) -> datetime.date:
703 """Deserializes string to date.
705 :param string: str.
706 :return: date.
707 """
708 try:
709 return parse(string).date()
710 except ImportError:
711 return string # type: ignore[return-value]
712 except ValueError as err:
713 raise ApiException(
714 status=0, reason=f'Failed to parse `{string}` as date object'
715 ) from err
717 def __deserialize_datetime(self, string: str) -> datetime.datetime:
718 """Deserializes string to datetime.
720 The string should be in iso8601 datetime format.
722 :param string: str.
723 :return: datetime.
724 """
725 try:
726 return parse(string)
727 except ImportError:
728 return string # type: ignore[return-value]
729 except ValueError as err:
730 raise ApiException(
731 status=0, reason=(f'Failed to parse `{string}` as datetime object')
732 ) from err
734 def __deserialize_enum(self, data: Any, klass: type[Enum]) -> Enum:
735 """Deserializes primitive type to enum.
737 :param data: primitive type.
738 :param klass: class literal.
739 :return: enum value.
740 """
741 try:
742 return klass(data)
743 except ValueError as err:
744 raise ApiException(
745 status=0, reason=(f'Failed to parse `{data}` as `{klass}`')
746 ) from err
748 def __deserialize_model(self, data: Any, klass: Any) -> Any:
749 """Deserializes list or dict to model.
751 :param data: dict, list.
752 :param klass: class literal.
753 :return: model object.
754 """
756 return klass.from_dict(data)