# coding: utf-8 """ TaskingAI OpenAPI spec version: 0.1.0 """ from __future__ import absolute_import import datetime import json import os import re import tempfile from typing import Type # python 2 and python 3 compatibility library import six from six.moves.urllib.parse import quote from taskingai.client.configuration import Configuration from taskingai.client import rest from pydantic import BaseModel class BaseApiClient(object): """Generic API client for Swagger client library builds. Swagger generic API client. This client handles the client- server communication, and is invariant across implementations. Specifics of the methods and models for each application are generated from the Swagger templates. NOTE: This class is auto generated by the swagger code generator program. Ref: https://github.com/swagger-api/swagger-codegen Do not edit the class manually. :param configuration: .Configuration object for this client :param header_name: a header to pass when making calls to the API. :param header_value: a header value to pass when making calls to the API. :param cookie: a cookie to include in the header when making calls to the API """ PRIMITIVE_TYPES = (float, bool, bytes, six.text_type) + six.integer_types NATIVE_TYPES_MAPPING = { "int": int, "long": int if six.PY3 else long, # noqa: F821 "float": float, "str": str, "bool": bool, "date": datetime.date, "datetime": datetime.datetime, "object": object, } def __init__(self, configuration=None, header_name=None, header_value=None, cookie=None): if configuration is None: configuration = Configuration() self.configuration = configuration self.default_headers = {} if header_name is not None: self.default_headers[header_name] = header_value self.cookie = cookie # Set default User-Agent. self.user_agent = "TaskingAI-Client/1.0.0/python" @property def user_agent(self): """User agent for this API client""" return self.default_headers["User-Agent"] @user_agent.setter def user_agent(self, value): self.default_headers["User-Agent"] = value def set_default_header(self, header_name, header_value): self.default_headers[header_name] = header_value def deserialize(self, response, response_type: Type[BaseModel]): """Deserializes response into an object. :param response: RESTResponse object to be deserialized. :param response_type: class literal for deserialized object, or string of class name. :return: deserialized object. """ # handle file downloading # save response body into a tmp file and return the instance if response_type == "file": return self.__deserialize_file(response) # fetch data from response object try: data = json.loads(response.data) except ValueError: data = response.data return response_type(**data) def select_header_accept(self, accepts): """Returns `Accept` based on an array of accepts provided. :param accepts: List of headers. :return: Accept (e.g. application/json). """ if not accepts: return accepts = [x.lower() for x in accepts] if "application/json" in accepts: return "application/json" else: return ", ".join(accepts) def select_header_content_type(self, content_types): """Returns `Content-Type` based on an array of content_types provided. :param content_types: List of content-types. :return: Content-Type (e.g. application/json). """ if not content_types: return "application/json" content_types = [x.lower() for x in content_types] if "application/json" in content_types or "*/*" in content_types: return "application/json" else: return content_types[0] def update_params_for_auth(self, headers, querys, auth_settings): """Updates header and query params based on authentication setting. :param headers: Header parameters dict to be updated. :param querys: Query parameters tuple list to be updated. :param auth_settings: Authentication setting identifiers list. """ if not auth_settings: return for auth in auth_settings: auth_setting = self.configuration.auth_settings().get(auth) if auth_setting: if not auth_setting["value"]: continue elif auth_setting["in"] == "header": headers[auth_setting["key"]] = auth_setting["value"] elif auth_setting["in"] == "query": querys.append((auth_setting["key"], auth_setting["value"])) else: raise ValueError("Authentication token must be in `query` or `header`") def __deserialize_file(self, response): """Deserializes body to file Saves response body into a file in a temporary folder, using the filename from the `Content-Disposition` header if provided. :param response: RESTResponse. :return: file path. """ fd, path = tempfile.mkstemp(dir=self.configuration.temp_folder_path) os.close(fd) os.remove(path) content_disposition = response.getheader("Content-Disposition") if content_disposition: filename = re.search(r'filename=[\'"]?([^\'"\s]+)[\'"]?', content_disposition).group(1) path = os.path.join(os.path.dirname(path), filename) response_data = response.data with open(path, "wb") as f: if isinstance(response_data, str): # change str to bytes so we can write it response_data = response_data.encode("utf-8") f.write(response_data) else: f.write(response_data) return path def __hasattr(self, object, name): return name in object.__class__.__dict__ class SyncApiClient(BaseApiClient): def __init__(self, *args, **kwargs): super(SyncApiClient, self).__init__(*args, **kwargs) self.rest_client = rest.RESTSyncClientObject(self.configuration) def __call_api( self, resource_path, method, path_params=None, query_params=None, header_params=None, body=None, post_params=None, files=None, response_type=None, auth_settings=None, _return_http_data_only=None, collection_formats=None, _preload_content=True, _request_timeout=None, stream=False, ): config = self.configuration # header parameters header_params = header_params or {} header_params.update(self.default_headers) if self.cookie: header_params["Cookie"] = self.cookie if header_params: if config.api_key_prefix and config.api_key: header_params["Authorization"] = f"{config.api_key_prefix} {config.api_key}" # path parameters if path_params: for k, v in path_params.items(): # specified safe chars, encode everything resource_path = resource_path.replace("{%s}" % k, quote(str(v), safe=config.safe_chars_for_path_param)) # auth setting self.update_params_for_auth(header_params, query_params, auth_settings) # body # request url url = self.configuration.host + resource_path # perform request and return response response_data = self.request( method, url, stream=stream, query_params=query_params, headers=header_params, post_params=post_params, files=files, body=body, _preload_content=_preload_content, _request_timeout=_request_timeout, ) if stream: return response_data self.last_response = response_data return_data = response_data if _preload_content: # deserialize response data if response_type: return_data = self.deserialize(response_data, response_type) else: return_data = None if _return_http_data_only: return return_data else: return (return_data, response_data.status, response_data.getheaders()) def call_api( self, resource_path, method, path_params=None, query_params=None, header_params=None, body=None, post_params=None, files=None, response_type=None, auth_settings=None, _return_http_data_only=None, collection_formats=None, _preload_content=True, _request_timeout=None, stream=False, ): """Makes the HTTP request (synchronous) and returns deserialized data. :param resource_path: Path to method endpoint. :param method: Method to call. :param path_params: Path parameters in the url. :param query_params: Query parameters in the url. :param header_params: Header parameters to be placed in the request header. :param body: Request body. :param post_params dict: Request post form parameters, for `application/x-www-form-urlencoded`, `multipart/form-data`. :param auth_settings list: Auth Settings names for the request. :param response: Response data type. :param files dict: key -> filename, value -> filepath, for `multipart/form-data`. :param _return_http_data_only: response data without head status code and headers :param collection_formats: dict of collection formats for path, query, header, and post parameters. :param _preload_content: if False, the urllib3.HTTPResponse object will be returned without reading/decoding response data. Default is True. :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :return: Response object data. """ return self.__call_api( resource_path, method, path_params, query_params, header_params, body, post_params, files, response_type, auth_settings, _return_http_data_only, collection_formats, _preload_content, _request_timeout, stream, ) def request( self, method, url, stream=False, query_params=None, headers=None, post_params=None, files=None, body=None, _preload_content=True, _request_timeout=None, ): """Makes the HTTP request using RESTClient.""" if method == "GET": return self.rest_client.GET( url, stream=stream, query_params=query_params, _preload_content=_preload_content, _request_timeout=_request_timeout, headers=headers, ) elif method == "HEAD": return self.rest_client.HEAD( url, stream=stream, query_params=query_params, _preload_content=_preload_content, _request_timeout=_request_timeout, headers=headers, ) elif method == "OPTIONS": return self.rest_client.OPTIONS( url, stream=stream, query_params=query_params, headers=headers, post_params=post_params, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body, ) elif method == "POST": return self.rest_client.POST( url, stream=stream, query_params=query_params, headers=headers, post_params=post_params, files=files, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body, ) elif method == "PUT": return self.rest_client.PUT( url, stream=stream, query_params=query_params, headers=headers, post_params=post_params, files=files, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body, ) elif method == "PATCH": return self.rest_client.PATCH( url, stream=stream, query_params=query_params, headers=headers, post_params=post_params, files=files, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body, ) elif method == "DELETE": return self.rest_client.DELETE( url, stream=stream, query_params=query_params, headers=headers, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body, ) else: raise ValueError("http method must be `GET`, `HEAD`, `OPTIONS`," " `POST`, `PATCH`, `PUT` or `DELETE`.") class AsyncApiClient(BaseApiClient): def __init__(self, *args, **kwargs): super(AsyncApiClient, self).__init__(*args, **kwargs) # Initialize the asynchronous REST client here self.rest_client = rest.RESTAsyncClientObject(self.configuration) async def __call_api( self, resource_path, method, path_params=None, query_params=None, header_params=None, body=None, post_params=None, files=None, response_type=None, auth_settings=None, _return_http_data_only=None, collection_formats=None, _preload_content=True, _request_timeout=None, stream=False, ): config = self.configuration # header parameters header_params = header_params or {} header_params.update(self.default_headers) if self.cookie: header_params["Cookie"] = self.cookie if header_params: if config.api_key_prefix and config.api_key: header_params["Authorization"] = f"{config.api_key_prefix} {config.api_key}" # path parameters if path_params: for k, v in path_params.items(): # specified safe chars, encode everything resource_path = resource_path.replace("{%s}" % k, quote(str(v), safe=config.safe_chars_for_path_param)) # auth setting self.update_params_for_auth(header_params, query_params, auth_settings) # body # request url url = self.configuration.host + resource_path # perform request and return response response_data = await self.request( method, url, stream=stream, query_params=query_params, headers=header_params, post_params=post_params, files=files, body=body, _preload_content=_preload_content, _request_timeout=_request_timeout, ) if stream: return response_data self.last_response = response_data return_data = response_data if _preload_content: # deserialize response data if response_type: return_data = self.deserialize(response_data, response_type) else: return_data = None if _return_http_data_only: return return_data else: return (return_data, response_data.status, response_data.getheaders()) async def call_api( self, resource_path, method, path_params=None, query_params=None, header_params=None, body=None, post_params=None, files=None, response_type=None, auth_settings=None, _return_http_data_only=None, collection_formats=None, _preload_content=True, _request_timeout=None, stream=False, ): """ Asynchronously makes the HTTP request and returns deserialized data. This method prepares and executes an HTTP request asynchronously based on the specified parameters and the method to be called. It supports various HTTP methods like GET, POST, PUT, DELETE, etc. :param resource_path: Path to the API endpoint. For example, '/items/{itemId}'. :param method: HTTP method to be used for the request, e.g., 'GET', 'POST', 'PUT', 'DELETE'. :param path_params: Dictionary of path parameters to be replaced in the resource_path. :param query_params: Dictionary of query parameters to be appended to the URL. :param header_params: Dictionary of header parameters to be included in the request header. :param body: Payload to be sent with the request. Usually for POST/PUT/PATCH methods. :param post_params: Dictionary of parameters for form-encoded data (used with POST, typically for file uploads). :param files: Dictionary with file names as keys and file paths as values for multipart file uploading. :param response_type: Expected data type of the response, e.g., a model class or a primitive type. :param auth_settings: List of authentication setting names to configure the request. :param _return_http_data_only: If True, returns only the data part of the response, excluding status code and headers. :param collection_formats: Specifies the format of collection parameters (query, path, header, post parameters). :param _preload_content: If False, the response object will be returned without reading/decoding response data. Default is True. :param _request_timeout: Specifies the request timeout. Can be a single number (total timeout) or a tuple (connection, read timeouts). :return: If _return_http_data_only is True, returns only the data part of the response. Otherwise, returns a tuple containing data, HTTP status code, and HTTP headers. """ return await self.__call_api( resource_path, method, path_params, query_params, header_params, body, post_params, files, response_type, auth_settings, _return_http_data_only, collection_formats, _preload_content, _request_timeout, stream, ) async def request( self, method, url, stream, query_params=None, headers=None, post_params=None, files=None, body=None, _preload_content=True, _request_timeout=None, ): """Makes the HTTP request using the asynchronous RESTClient.""" if method == "GET": return await self.rest_client.GET( url, stream=stream, query_params=query_params, _preload_content=_preload_content, _request_timeout=_request_timeout, headers=headers, ) elif method == "HEAD": return await self.rest_client.HEAD( url, stream=stream, query_params=query_params, _preload_content=_preload_content, _request_timeout=_request_timeout, headers=headers, ) elif method == "OPTIONS": return await self.rest_client.OPTIONS( url, stream=stream, query_params=query_params, headers=headers, post_params=post_params, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body, ) elif method == "POST": return await self.rest_client.POST( url, stream=stream, query_params=query_params, headers=headers, post_params=post_params, files=files, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body, ) elif method == "PUT": return await self.rest_client.PUT( url, stream=stream, query_params=query_params, headers=headers, post_params=post_params, files=files, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body, ) elif method == "PATCH": return await self.rest_client.PATCH( url, stream=stream, query_params=query_params, headers=headers, post_params=post_params, files=files, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body, ) elif method == "DELETE": return await self.rest_client.DELETE( url, stream=stream, query_params=query_params, headers=headers, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body, ) else: raise ValueError("HTTP method must be `GET`, `HEAD`, `OPTIONS`, " "`POST`, `PATCH`, `PUT`, or `DELETE`.")