api_client.py

# 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`.")