api_client.py

"""
Python SDK for OpenFGA

API version: 1.x
Website: https://openfga.dev
Documentation: https://openfga.dev/docs
Support: https://openfga.dev/community
License: [Apache-2.0](https://github.com/openfga/python-sdk/blob/main/LICENSE)

NOTE: This file was auto generated by OpenAPI Generator (https://openapi-generator.tech). DO NOT EDIT.
"""

import asyncio
import atexit
import datetime
import json
import math
import random
import re
import time
import urllib

from multiprocessing.pool import ThreadPool

from dateutil.parser import parse  # type: ignore[import-untyped]

import openfga_sdk.models

from openfga_sdk import oauth2, rest
from openfga_sdk.configuration import Configuration
from openfga_sdk.exceptions import (
    ApiException,
    ApiValueError,
    FgaValidationException,
    RateLimitExceededError,
    ServiceException,
)
from openfga_sdk.telemetry import Telemetry
from openfga_sdk.telemetry.attributes import TelemetryAttribute, TelemetryAttributes


DEFAULT_USER_AGENT = "openfga-sdk python/0.9.6"


def random_time(loop_count, min_wait_in_ms) -> float:
    """
    Helper function to return the time (in s) to wait before retry
    """
    minimum = math.ceil(2**loop_count * min_wait_in_ms)
    maximum = math.ceil(2 ** (loop_count + 1) * min_wait_in_ms)
    return random.randrange(minimum, maximum) / 1000


class ApiClient:
    """Generic API client for OpenAPI client library builds.

    OpenAPI 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 OpenAPI
    templates.

    NOTE: This class is auto generated by OpenAPI Generator.
    Ref: https://openapi-generator.tech
    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
    :param pool_threads: The number of threads to use for async requests
        to the API. More threads means more concurrent API requests.
    """

    PRIMITIVE_TYPES = (float, bool, bytes, str, int)
    NATIVE_TYPES_MAPPING = {
        "int": int,
        "long": int,
        "float": float,
        "str": str,
        "bool": bool,
        "date": datetime.date,
        "datetime": datetime.datetime,
        "object": object,
    }
    _pool = None

    def __init__(
        self,
        configuration=None,
        header_name=None,
        header_value=None,
        cookie=None,
        pool_threads=1,
    ):
        if configuration is None:
            configuration = Configuration.get_default_copy()

        self.configuration = configuration
        self.pool_threads = pool_threads

        self.rest_client = rest.RESTClientObject(configuration)

        self.default_headers = {}
        if header_name is not None:
            self.default_headers[header_name] = header_value

        self.cookie = cookie

        self.user_agent = DEFAULT_USER_AGENT

        self.client_side_validation = configuration.client_side_validation
        self._telemetry = Telemetry()

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_value, traceback):
        await self.close()

    async def close(self):
        await self.rest_client.close()
        if self._pool:
            self._pool.close()
            self._pool.join()
            self._pool = None
            if hasattr(atexit, "unregister"):
                atexit.unregister(self.close)

    @property
    def pool(self):
        """Create thread pool on first request
        avoids instantiating unused threadpool for blocking clients.
        """
        if self._pool is None:
            atexit.register(self.close)
            self._pool = ThreadPool(self.pool_threads)
        return self._pool

    @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

    async def __call_api(
        self,
        resource_path,
        method,
        path_params=None,
        query_params=None,
        header_params=None,
        body=None,
        post_params=None,
        response_types_map=None,
        auth_settings=None,
        _return_http_data_only=None,
        collection_formats=None,
        _preload_content=True,
        _request_timeout=None,
        _host=None,
        _request_auth=None,
        _retry_params=None,
        _oauth2_client=None,
        _telemetry_attributes: dict[TelemetryAttribute, str | bool | int | float]
        | None = None,
        _streaming: bool = False,
    ):
        self.configuration.is_valid()
        config = self.configuration
        start = float(time.time())

        # header parameters
        header_params = header_params or {}
        header_params.update(self.default_headers)
        if self.cookie:
            header_params["Cookie"] = self.cookie
        if header_params:
            header_params = self.sanitize_for_serialization(header_params)
            header_params = dict(
                self.parameters_to_tuples(header_params, collection_formats)
            )

        # path parameters
        if path_params:
            path_params = self.sanitize_for_serialization(path_params)
            path_params = self.parameters_to_tuples(path_params, collection_formats)
            for k, v in path_params:
                # specified safe chars, encode everything
                _k = urllib.parse.quote(str(k), safe=config.safe_chars_for_path_param)
                _v = urllib.parse.quote(str(v), safe=config.safe_chars_for_path_param)
                resource_path = resource_path.replace("{" + str(k) + "}", _v)

        # query parameters
        if query_params:
            query_params = self.sanitize_for_serialization(query_params)
            query_params = self.parameters_to_tuples(query_params, collection_formats)

        # post parameters
        if post_params:
            post_params = post_params if post_params else []
            post_params = self.sanitize_for_serialization(post_params)
            post_params = self.parameters_to_tuples(post_params, collection_formats)

        # auth setting
        await self.update_params_for_auth(
            header_params,
            query_params,
            auth_settings,
            request_auth=_request_auth,
            oauth2_client=_oauth2_client,
        )

        # body
        if body:
            body = self.sanitize_for_serialization(body)

        # request url
        if _host is None:
            if self.configuration.api_url is not None:
                url = self.configuration.api_url + resource_path
            else:
                url = (
                    self.configuration.api_scheme
                    + "://"
                    + self.configuration.api_host
                    + resource_path
                )
        else:
            # use server/host defined in path or operation instead
            url = self.configuration.api_scheme + "://" + _host + resource_path

        max_retry = (
            self.configuration.retry_params.max_retry
            if (
                self.configuration.retry_params is not None
                and self.configuration.retry_params.max_retry is not None
            )
            else 0
        )
        min_wait_in_ms = (
            self.configuration.retry_params.min_wait_in_ms
            if (
                self.configuration.retry_params is not None
                and self.configuration.retry_params.min_wait_in_ms is not None
            )
            else 0
        )
        max_wait_in_sec = (
            self.configuration.retry_params.max_wait_in_sec
            if (
                self.configuration.retry_params is not None
                and self.configuration.retry_params.max_wait_in_sec is not None
            )
            else 120
        )
        if _retry_params is not None:
            if _retry_params.max_retry is not None:
                max_retry = _retry_params.max_retry
            if _retry_params.min_wait_in_ms is not None:
                max_retry = _retry_params.min_wait_in_ms
            if _retry_params.max_wait_in_sec is not None:
                max_wait_in_sec = _retry_params.max_wait_in_sec

        _telemetry_attributes = TelemetryAttributes.fromRequest(
            user_agent=self.user_agent,
            fga_method=resource_path,
            http_method=method,
            url=url,
            resend_count=0,
            start=start,
            credentials=self.configuration.credentials,
            attributes=_telemetry_attributes,
        )

        for retry in range(max_retry + 1):
            _telemetry_attributes[TelemetryAttributes.http_request_resend_count] = retry

            try:
                # perform request and return response
                response_data = await self.request(
                    method,
                    url,
                    query_params=query_params,
                    headers=header_params,
                    post_params=post_params,
                    body=body,
                    _preload_content=_preload_content,
                    _request_timeout=_request_timeout,
                    _streaming=_streaming,
                )
            except (RateLimitExceededError, ServiceException) as e:
                if retry < max_retry and e.status != 501:
                    _telemetry_attributes = TelemetryAttributes.fromResponse(
                        response=e.body.decode("utf-8"),
                        credentials=self.configuration.credentials,
                        attributes=_telemetry_attributes,
                        start=start,
                    )

                    self._telemetry.metrics.request(
                        attributes=_telemetry_attributes,
                        configuration=self.configuration.telemetry,
                    )

                    try:
                        wait_time_in_sec = self._parse_retry_after_header(e.header)
                    except ValueError:
                        wait_time_in_sec = min(
                            random_time(retry, min_wait_in_ms), max_wait_in_sec
                        )

                    await asyncio.sleep(wait_time_in_sec)

                    continue
                e.body = e.body.decode("utf-8")
                response_type = response_types_map.get(e.status, None)
                if response_type is not None:
                    e.parsed_exception = self.__deserialize(
                        json.loads(e.body), response_type
                    )
                    e.body = None
                raise e
            except ApiException as e:
                e.body = e.body.decode("utf-8")
                response_type = response_types_map.get(e.status, None)
                if response_type is not None:
                    e.parsed_exception = self.__deserialize(
                        json.loads(e.body), response_type
                    )
                    e.body = None

                _telemetry_attributes = TelemetryAttributes.fromResponse(
                    response=e,
                    credentials=self.configuration.credentials,
                    attributes=_telemetry_attributes,
                    start=start,
                )

                self._telemetry.metrics.request(
                    attributes=_telemetry_attributes,
                    configuration=self.configuration.telemetry,
                )

                self._telemetry.metrics.queryDuration(
                    attributes=_telemetry_attributes,
                    configuration=self.configuration.telemetry,
                )

                self._telemetry.metrics.requestDuration(
                    attributes=_telemetry_attributes,
                    configuration=self.configuration.telemetry,
                )
                raise e

            self.last_response = response_data

            return_data = response_data

            _telemetry_attributes = TelemetryAttributes.fromResponse(
                response=response_data,
                credentials=self.configuration.credentials,
                attributes=_telemetry_attributes,
                start=start,
            )

            self._telemetry.metrics.request(
                attributes=_telemetry_attributes,
                configuration=self.configuration.telemetry,
            )

            self._telemetry.metrics.queryDuration(
                attributes=_telemetry_attributes,
                configuration=self.configuration.telemetry,
            )

            self._telemetry.metrics.requestDuration(
                attributes=_telemetry_attributes,
                configuration=self.configuration.telemetry,
            )

            if not _preload_content or _streaming:
                return return_data

            response_type = response_types_map.get(response_data.status, None)

            if response_type not in ["file", "bytes"]:
                match = None
                content_type = response_data.getheader("content-type")
                if content_type is not None:
                    match = re.search(r"charset=([a-zA-Z\-\d]+)[\s\;]?", content_type)
                encoding = match.group(1) if match else "utf-8"
                if response_data.data is not None:
                    response_data.data = response_data.data.decode(encoding)

            # 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.headers)

    def _parse_retry_after_header(self, headers) -> int:
        retry_after_header = headers.get("retry-after")
        if not retry_after_header:
            raise ValueError("Retry-After header is not present")

        try:
            parsed_http_date = self.__deserialize_datetime(retry_after_header).replace(
                tzinfo=datetime.timezone.utc
            )
            now = datetime.datetime.now(datetime.timezone.utc)
            wait_time_in_sec = (parsed_http_date - now).total_seconds()
        except ApiException:
            wait_time_in_sec = int(retry_after_header)

        if wait_time_in_sec > 1800 or wait_time_in_sec < 1:
            raise ValueError("Retry-After header is invalid")

        return math.ceil(wait_time_in_sec)

    def sanitize_for_serialization(self, obj):
        """Builds a JSON POST object.

        If obj is None, return None.
        If obj is str, int, long, float, bool, return directly.
        If obj is datetime.datetime, datetime.date
            convert to string in iso8601 format.
        If obj is list, sanitize each element in the list.
        If obj is dict, return the dict.
        If obj is OpenAPI model, return the properties dict.

        :param obj: The data to serialize.
        :return: The serialized form of data.
        """
        if obj is None:
            return None
        elif isinstance(obj, self.PRIMITIVE_TYPES):
            return obj
        elif isinstance(obj, list):
            return [self.sanitize_for_serialization(sub_obj) for sub_obj in obj]
        elif isinstance(obj, tuple):
            return tuple(self.sanitize_for_serialization(sub_obj) for sub_obj in obj)
        elif isinstance(obj, datetime.datetime | datetime.date):
            return obj.isoformat()

        if isinstance(obj, dict):
            obj_dict = obj
        else:
            # Convert model obj to dict except
            # attributes `openapi_types`, `attribute_map`
            # and attributes which value is not None.
            # Convert attribute name to json key in
            # model definition for request.
            obj_dict = {
                obj.attribute_map[attr]: getattr(obj, attr)
                for attr, _ in obj.openapi_types.items()
                if getattr(obj, attr) is not None
            }

        return {
            key: self.sanitize_for_serialization(val) for key, val in obj_dict.items()
        }

    def deserialize(self, response, response_type):
        """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.
        """

        # fetch data from response object
        try:
            data = json.loads(response.data)
        except ValueError:
            data = response.data

        return self.__deserialize(data, response_type)

    def __deserialize(self, data, klass):
        """Deserializes dict, list, str into an object.

        :param data: dict, list or str.
        :param klass: class literal, or string of class name.

        :return: object.
        """
        if data is None:
            return None

        if type(klass) is str:
            if klass.startswith("list["):
                sub_kls = re.match(r"list\[(.*)\]", klass).group(1)
                return [self.__deserialize(sub_data, sub_kls) for sub_data in data]

            if klass.startswith("dict["):
                sub_kls = re.match(r"dict\[([^,]*), (.*)\]", klass).group(2)
                return {k: self.__deserialize(v, sub_kls) for k, v in data.items()}

            # convert str to class
            if klass in self.NATIVE_TYPES_MAPPING:
                klass = self.NATIVE_TYPES_MAPPING[klass]
            else:
                klass = getattr(openfga_sdk.models, klass)

        if klass in self.PRIMITIVE_TYPES:
            return self.__deserialize_primitive(data, klass)

        if klass is object:
            return self.__deserialize_object(data)

        if klass is datetime.date:
            return self.__deserialize_date(data)

        if klass is datetime.datetime:
            return self.__deserialize_datetime(data)

        return self.__deserialize_model(data, klass)

    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_types_map=None,
        auth_settings=None,
        async_req=None,
        _return_http_data_only=None,
        collection_formats=None,
        _preload_content=True,
        _request_timeout=None,
        _host=None,
        _request_auth=None,
        _retry_params=None,
        _oauth2_client=None,
        _telemetry_attributes: dict[TelemetryAttribute, str | bool | int | float]
        | None = None,
        _streaming: bool = False,
    ):
        """Makes the HTTP request (synchronous) and returns deserialized data.

        To make an async_req request, set the async_req parameter.

        :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: it will not be used
        :param async_req bool: execute request asynchronously
        :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.
        :param _request_auth: set to override the auth_settings for an a single
                              request; this effectively ignores the authentication
                              in the spec for a single request.
        :param _retry_params: If specified, override the default retry parameters
        :type _request_token: dict, optional
        :return:
            If async_req parameter is True,
            the request will be called asynchronously.
            The method will return the request thread.
            If parameter async_req is False or missing,
            then the method will return the response directly.
        """
        if not async_req:
            return await self.__call_api(
                resource_path,
                method,
                path_params,
                query_params,
                header_params,
                body,
                post_params,
                response_types_map,
                auth_settings,
                _return_http_data_only,
                collection_formats,
                _preload_content,
                _request_timeout,
                _host,
                _request_auth,
                _retry_params,
                _oauth2_client,
                _telemetry_attributes,
                _streaming,
            )

        return self.pool.apply_async(
            self.__call_api,
            (
                resource_path,
                method,
                path_params,
                query_params,
                header_params,
                body,
                post_params,
                response_types_map,
                auth_settings,
                _return_http_data_only,
                collection_formats,
                _preload_content,
                _request_timeout,
                _host,
                _request_auth,
                _retry_params,
                _oauth2_client,
                _telemetry_attributes,
                _streaming,
            ),
        )

    async def request(
        self,
        method,
        url,
        query_params=None,
        headers=None,
        post_params=None,
        body=None,
        _preload_content=True,
        _request_timeout=None,
        _streaming: bool = False,
    ):
        if method not in ["GET", "HEAD", "OPTIONS", "POST", "PATCH", "PUT", "DELETE"]:
            raise ApiValueError(
                "http method must be `GET`, `HEAD`, `OPTIONS`,"
                " `POST`, `PATCH`, `PUT` or `DELETE`."
            )

        if _streaming:
            return self.rest_client.stream(
                method,
                url,
                query_params=query_params,
                headers=headers,
                post_params=post_params,
                body=body,
                _request_timeout=_request_timeout,
            )

        return await self.rest_client.request(
            method,
            url,
            query_params=query_params,
            headers=headers,
            post_params=post_params,
            body=body,
            _preload_content=_preload_content,
            _request_timeout=_request_timeout,
        )

    def parameters_to_tuples(self, params, collection_formats):
        """Get parameters as list of tuples, formatting collections.

        :param params: Parameters as dict or list of two-tuples
        :param dict collection_formats: Parameter collection formats
        :return: Parameters as list of tuples, collections formatted
        """
        new_params = []
        if collection_formats is None:
            collection_formats = {}
        for k, v in params.items() if isinstance(params, dict) else params:
            if k in collection_formats:
                collection_format = collection_formats[k]
                if collection_format == "multi":
                    new_params.extend((k, value) for value in v)
                else:
                    if collection_format == "ssv":
                        delimiter = " "
                    elif collection_format == "tsv":
                        delimiter = "\t"
                    elif collection_format == "pipes":
                        delimiter = "|"
                    else:  # csv is the default
                        delimiter = ","
                    new_params.append((k, delimiter.join(str(value) for value in v)))
            else:
                new_params.append((k, v))
        return new_params

    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, method=None, body=None):
        """Returns `Content-Type` based on an array of content_types provided.

        :param content_types: List of content-types.
        :param method: http method (e.g. POST, PATCH).
        :param body: http body to send.
        :return: Content-Type (e.g. application/json).
        """
        if not content_types:
            return None

        content_types = [x.lower() for x in content_types]

        if (
            method == "PATCH"
            and "application/json-patch+json" in content_types
            and isinstance(body, list)
        ):
            return "application/json-patch+json"

        if "application/json" in content_types or "*/*" in content_types:
            return "application/json"
        else:
            return content_types[0]

    async def update_params_for_auth(
        self, headers, queries, auth_settings, request_auth=None, oauth2_client=None
    ):
        """Updates header and query params based on authentication setting.

        :param headers: Header parameters dict to be updated.
        :param queries: Query parameters tuple list to be updated.
        :param auth_settings: Authentication setting identifiers list.
        :param request_auth: if set, the provided settings will
                             override the token in the configuration.
        :param oauth2_client: if set, will be used for credential exchange.
        """
        credentials = self.configuration.credentials
        if credentials is not None:
            if credentials.method == "none":
                pass
            if credentials.method == "api_token":
                headers["Authorization"] = (
                    f"Bearer {credentials.configuration.api_token}"
                )
            if credentials.method == "client_credentials":
                if oauth2_client is None:
                    oauth2_client = oauth2.OAuth2Client(credentials, self.configuration)
                oauth2_headers = await oauth2_client.get_authentication_header(
                    self.rest_client
                )
                for key, value in oauth2_headers.items():
                    headers[key] = value

        if not auth_settings:
            return

        if request_auth:
            self._apply_auth_params(headers, queries, request_auth)
            return

        for auth in auth_settings:
            auth_setting = self.configuration.auth_settings().get(auth)
            if auth_setting:
                self._apply_auth_params(headers, queries, auth_setting)

    def _apply_auth_params(self, headers, queries, auth_setting):
        """Updates the request parameters based on a single auth_setting

        :param headers: Header parameters dict to be updated.
        :param queries: Query parameters tuple list to be updated.
        :param auth_setting: auth settings for the endpoint
        """
        if auth_setting["in"] == "cookie":
            headers["Cookie"] = auth_setting["value"]
        elif auth_setting["in"] == "header":
            headers[auth_setting["key"]] = auth_setting["value"]
        elif auth_setting["in"] == "query":
            queries.append((auth_setting["key"], auth_setting["value"]))
        else:
            raise ApiValueError("Authentication token must be in `query` or `header`")

    def __deserialize_primitive(self, data, klass):
        """Deserializes string to primitive type.

        :param data: str.
        :param klass: class literal.

        :return: int, long, float, str, bool.
        """
        try:
            return klass(data)
        except UnicodeEncodeError:
            return str(data)
        except TypeError:
            return data

    def __deserialize_object(self, value):
        """Return an original value.

        :return: object.
        """
        return value

    def __deserialize_date(self, string):
        """Deserializes string to date.

        :param string: str.
        :return: date.
        """
        try:
            return parse(string).date()
        except ImportError:
            return string
        except ValueError:
            raise rest.ApiException(
                status=0, reason=f"Failed to parse `{string}` as date object"
            )

    def __deserialize_datetime(self, string):
        """Deserializes string to datetime.

        The string should be in iso8601 datetime format.

        :param string: str.
        :return: datetime.
        """
        try:
            return parse(string)
        except ImportError:
            return string
        except (TypeError, ValueError):
            raise rest.ApiException(
                status=0,
                reason=(f"Failed to parse `{string}` as datetime object"),
            )

    def __deserialize_model(self, data, klass):
        """Deserializes list or dict to model.

        :param data: dict, list.
        :param klass: class literal.
        :return: model object.
        """
        has_discriminator = False
        if (
            hasattr(klass, "get_real_child_model")
            and klass.discriminator_value_class_map
        ):
            has_discriminator = True

        if not klass.openapi_types and has_discriminator is False:
            return data

        kwargs = {}
        if (
            data is not None
            and klass.openapi_types is not None
            and isinstance(data, list | dict)
        ):
            for attr, attr_type in klass.openapi_types.items():
                if klass.attribute_map[attr] in data:
                    value = data[klass.attribute_map[attr]]
                    kwargs[attr] = self.__deserialize(value, attr_type)

        kwargs["local_vars_configuration"] = self.configuration
        instance = klass(**kwargs)

        if has_discriminator:
            klass_name = instance.get_real_child_model(data)
            if klass_name:
                instance = self.__deserialize(data, klass_name)
        return instance

    def _get_store_id(self):
        """
        Verify that the store id has been configured and not empty string.
        It will return the store ID.
        Otherwise, raise FgaValidationException
        """
        configuration = self.configuration
        if configuration.store_id is None or configuration.store_id == "":
            raise FgaValidationException("store_id is required but not configured")
        return configuration.store_id

    def set_store_id(self, value):
        """
        Update the store ID in the configuration
        """
        self.configuration.store_id = value

    def get_store_id(self):
        """
        Return the store id (if any) store in the configuration
        """
        return self.configuration.store_id