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