api_client.handlebars

# coding: utf-8
{{>partial_header}}

import datetime
import dataclasses
import decimal
import enum
import email
import json
import os
import io
import atexit
from multiprocessing import pool
import re
import tempfile
import typing
import typing_extensions
from urllib import parse
import urllib3
from urllib3 import _collections, fields

{{#if tornado}}
import tornado.gen
{{/if}}
import frozendict

from {{packageName}} import exceptions, rest, schemas
from {{packageName}} import configuration as configuration_module


class RequestField(fields.RequestField):
    def __eq__(self, other):
        if not isinstance(other, RequestField):
            return False
        return self.__dict__ == other.__dict__


class JSONEncoder(json.JSONEncoder):
    compact_separators = (',', ':')

    def default(self, obj):
        if isinstance(obj, str):
            return str(obj)
        elif isinstance(obj, float):
            return float(obj)
        elif isinstance(obj, int):
            return int(obj)
        elif isinstance(obj, decimal.Decimal):
            if obj.as_tuple().exponent >= 0:
                return int(obj)
            return float(obj)
        elif isinstance(obj, schemas.NoneClass):
            return None
        elif isinstance(obj, schemas.BoolClass):
            return bool(obj)
        elif isinstance(obj, (dict, frozendict.frozendict)):
            return {key: self.default(val) for key, val in obj.items()}
        elif isinstance(obj, (list, tuple)):
            return [self.default(item) for item in obj]
        raise exceptions.ApiValueError('Unable to prepare type {} for serialization'.format(obj.__class__.__name__))


class ParameterInType(enum.Enum):
    QUERY = 'query'
    HEADER = 'header'
    PATH = 'path'
    COOKIE = 'cookie'


class ParameterStyle(enum.Enum):
    MATRIX = 'matrix'
    LABEL = 'label'
    FORM = 'form'
    SIMPLE = 'simple'
    SPACE_DELIMITED = 'spaceDelimited'
    PIPE_DELIMITED = 'pipeDelimited'
    DEEP_OBJECT = 'deepObject'


class PrefixSeparatorIterator:
    # A class to store prefixes and separators for rfc6570 expansions

    def __init__(self, prefix: str, separator: str):
        self.prefix = prefix
        self.separator = separator
        self.first = True
        if separator in {'.', '|', '%20'}:
            item_separator = separator
        else:
            item_separator = ','
        self.item_separator = item_separator

    def __iter__(self):
        return self

    def __next__(self):
        if self.first:
            self.first = False
            return self.prefix
        return self.separator


class ParameterSerializerBase:
    @staticmethod
    def __ref6570_item_value(in_data: typing.Any, percent_encode: bool):
        """
        Get representation if str/float/int/None/items in list/ values in dict
        None is returned if an item is undefined, use cases are value=
        - None
        - []
        - {}
        - [None, None None]
        - {'a': None, 'b': None}
        """
        if type(in_data) in {str, float, int}:
            if percent_encode:
                return parse.quote(str(in_data))
            return str(in_data)
        elif isinstance(in_data, schemas.none_type):
            # ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
            return None
        elif isinstance(in_data, list) and not in_data:
            # ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
            return None
        elif isinstance(in_data, dict) and not in_data:
            # ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
            return None
        raise exceptions.ApiValueError('Unable to generate a ref6570 item representation of {}'.format(in_data))

    @staticmethod
    def _to_dict(name: str, value: str):
        return {name: value}

    @classmethod
    def __ref6570_str_float_int_expansion(
        cls,
        variable_name: str,
        in_data: typing.Any,
        explode: bool,
        percent_encode: bool,
        prefix_separator_iterator: PrefixSeparatorIterator,
        var_name_piece: str,
        named_parameter_expansion: bool
    ) -> str:
        item_value = cls.__ref6570_item_value(in_data, percent_encode)
        if item_value is None or (item_value == '' and prefix_separator_iterator.separator == ';'):
            return next(prefix_separator_iterator) + var_name_piece
        value_pair_equals = '=' if named_parameter_expansion else ''
        return next(prefix_separator_iterator) + var_name_piece + value_pair_equals + item_value

    @classmethod
    def __ref6570_list_expansion(
        cls,
        variable_name: str,
        in_data: typing.Any,
        explode: bool,
        percent_encode: bool,
        prefix_separator_iterator: PrefixSeparatorIterator,
        var_name_piece: str,
        named_parameter_expansion: bool
    ) -> str:
        item_values = [cls.__ref6570_item_value(v, percent_encode) for v in in_data]
        item_values = [v for v in item_values if v is not None]
        if not item_values:
            # ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
            return ""
        value_pair_equals = '=' if named_parameter_expansion else ''
        if not explode:
            return (
                next(prefix_separator_iterator) +
                var_name_piece +
                value_pair_equals +
                prefix_separator_iterator.item_separator.join(item_values)
            )
        # exploded
        return next(prefix_separator_iterator) + next(prefix_separator_iterator).join(
            [var_name_piece + value_pair_equals + val for val in item_values]
        )

    @classmethod
    def __ref6570_dict_expansion(
        cls,
        variable_name: str,
        in_data: typing.Any,
        explode: bool,
        percent_encode: bool,
        prefix_separator_iterator: PrefixSeparatorIterator,
        var_name_piece: str,
        named_parameter_expansion: bool
    ) -> str:
        in_data_transformed = {key: cls.__ref6570_item_value(val, percent_encode) for key, val in in_data.items()}
        in_data_transformed = {key: val for key, val in in_data_transformed.items() if val is not None}
        if not in_data_transformed:
            # ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
            return ""
        value_pair_equals = '=' if named_parameter_expansion else ''
        if not explode:
            return (
                next(prefix_separator_iterator) +
                var_name_piece + value_pair_equals +
                prefix_separator_iterator.item_separator.join(
                    prefix_separator_iterator.item_separator.join(
                        item_pair
                    ) for item_pair in in_data_transformed.items()
                )
            )
        # exploded
        return next(prefix_separator_iterator) + next(prefix_separator_iterator).join(
            [key + '=' + val for key, val in in_data_transformed.items()]
        )

    @classmethod
    def _ref6570_expansion(
        cls,
        variable_name: str,
        in_data: typing.Any,
        explode: bool,
        percent_encode: bool,
        prefix_separator_iterator: PrefixSeparatorIterator
    ) -> str:
        """
        Separator is for separate variables like dict with explode true, not for array item separation
        """
        named_parameter_expansion = prefix_separator_iterator.separator in {'&', ';'}
        var_name_piece = variable_name if named_parameter_expansion else ''
        if type(in_data) in {str, float, int}:
            return cls.__ref6570_str_float_int_expansion(
                variable_name,
                in_data,
                explode,
                percent_encode,
                prefix_separator_iterator,
                var_name_piece,
                named_parameter_expansion
            )
        elif isinstance(in_data, schemas.none_type):
            # ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
            return ""
        elif isinstance(in_data, list):
            return cls.__ref6570_list_expansion(
                variable_name,
                in_data,
                explode,
                percent_encode,
                prefix_separator_iterator,
                var_name_piece,
                named_parameter_expansion
            )
        elif isinstance(in_data, dict):
            return cls.__ref6570_dict_expansion(
                variable_name,
                in_data,
                explode,
                percent_encode,
                prefix_separator_iterator,
                var_name_piece,
                named_parameter_expansion
            )
        # bool, bytes, etc
        raise exceptions.ApiValueError('Unable to generate a ref6570 representation of {}'.format(in_data))


class StyleFormSerializer(ParameterSerializerBase):
    @classmethod
    def _serialize_form(
        cls,
        in_data: typing.Union[None, int, float, str, bool, dict, list],
        name: str,
        explode: bool,
        percent_encode: bool,
        prefix_separator_iterator: typing.Optional[PrefixSeparatorIterator] = None
    ) -> str:
        if prefix_separator_iterator is None:
            prefix_separator_iterator = PrefixSeparatorIterator('', '&')
        return cls._ref6570_expansion(
            variable_name=name,
            in_data=in_data,
            explode=explode,
            percent_encode=percent_encode,
            prefix_separator_iterator=prefix_separator_iterator
        )


class StyleSimpleSerializer(ParameterSerializerBase):

    @classmethod
    def _serialize_simple(
        cls,
        in_data: typing.Union[None, int, float, str, bool, dict, list],
        name: str,
        explode: bool,
        percent_encode: bool
    ) -> str:
        prefix_separator_iterator = PrefixSeparatorIterator('', ',')
        return cls._ref6570_expansion(
            variable_name=name,
            in_data=in_data,
            explode=explode,
            percent_encode=percent_encode,
            prefix_separator_iterator=prefix_separator_iterator
        )

    @classmethod
    def _deserialize_simple(
        cls,
        in_data: str,
        name: str,
        explode: bool,
        percent_encode: bool
    ) -> typing.Union[str, typing.List[str], typing.Dict[str, str]]:
        raise NotImplementedError(
            "Deserialization of style=simple has not yet been added. "
            "If you need this how about you submit a PR adding it?"
        )


class JSONDetector:
    """
    Works for:
    application/json
    application/json; charset=UTF-8
    application/json-patch+json
    application/geo+json
    """
    __json_content_type_pattern = re.compile("application/[^+]*[+]?(json);?.*")

    @classmethod
    def _content_type_is_json(cls, content_type: str) -> bool:
        if cls.__json_content_type_pattern.match(content_type):
            return True
        return False


@dataclasses.dataclass
class ParameterBase(JSONDetector):
    in_type: ParameterInType
    required: bool
    style: typing.Optional[ParameterStyle]
    explode: typing.Optional[bool]
    allow_reserved: typing.Optional[bool]
    schema: typing.Optional[typing.Type[schemas.Schema]]
    content: typing.Optional[typing.Dict[str, typing.Type[schemas.Schema]]]

    _json_encoder = JSONEncoder()

    def __init_subclass__(cls, **kwargs):
        if cls.explode is None:
            if cls.style is ParameterStyle.FORM:
                cls.explode = True
            else:
                cls.explode = False

    @classmethod
    def _serialize_json(
        cls,
        in_data: typing.Union[None, int, float, str, bool, dict, list],
        eliminate_whitespace: bool = False
    ) -> str:
        if eliminate_whitespace:
            return json.dumps(in_data, separators=cls._json_encoder.compact_separators)
        return json.dumps(in_data)


class PathParameter(ParameterBase, StyleSimpleSerializer):
    name: str
    required: bool = False
    in_type: ParameterInType = ParameterInType.PATH
    style: ParameterStyle = ParameterStyle.SIMPLE
    explode: bool = False
    allow_reserved: typing.Optional[bool] = None
    schema: typing.Optional[typing.Type[schemas.Schema]] = None
    content: typing.Optional[typing.Dict[str, typing.Type[schemas.Schema]]] = None

    @classmethod
    def __serialize_label(
        cls,
        in_data: typing.Union[None, int, float, str, bool, dict, list]
    ) -> typing.Dict[str, str]:
        prefix_separator_iterator = PrefixSeparatorIterator('.', '.')
        value = cls._ref6570_expansion(
            variable_name=cls.name,
            in_data=in_data,
            explode=cls.explode,
            percent_encode=True,
            prefix_separator_iterator=prefix_separator_iterator
        )
        return cls._to_dict(cls.name, value)

    @classmethod
    def __serialize_matrix(
        cls,
        in_data: typing.Union[None, int, float, str, bool, dict, list]
    ) -> typing.Dict[str, str]:
        prefix_separator_iterator = PrefixSeparatorIterator(';', ';')
        value = cls._ref6570_expansion(
            variable_name=cls.name,
            in_data=in_data,
            explode=cls.explode,
            percent_encode=True,
            prefix_separator_iterator=prefix_separator_iterator
        )
        return cls._to_dict(cls.name, value)

    @classmethod
    def __serialize_simple(
        cls,
        in_data: typing.Union[None, int, float, str, bool, dict, list],
    ) -> typing.Dict[str, str]:
        value = cls._serialize_simple(
            in_data=in_data,
            name=cls.name,
            explode=cls.explode,
            percent_encode=True
        )
        return cls._to_dict(cls.name, value)

    @classmethod
    def serialize(
        cls,
        in_data: typing.Union[
            schemas.Schema, decimal.Decimal, int, float, str, datetime.date, datetime.datetime, None, bool, list, tuple, dict, frozendict.frozendict]
    ) -> typing.Dict[str, str]:
        if cls.schema:
            cast_in_data = cls.schema(in_data)
            cast_in_data = cls._json_encoder.default(cast_in_data)
            """
            simple -> path
                path:
                    returns path_params: dict
            label -> path
                returns path_params
            matrix -> path
                returns path_params
            """
            if cls.style:
                if cls.style is ParameterStyle.SIMPLE:
                    return cls.__serialize_simple(cast_in_data)
                elif cls.style is ParameterStyle.LABEL:
                    return cls.__serialize_label(cast_in_data)
                elif cls.style is ParameterStyle.MATRIX:
                    return cls.__serialize_matrix(cast_in_data)
        # cls.content will be length one
        for content_type, schema in cls.content.items():
            cast_in_data = schema(in_data)
            cast_in_data = cls._json_encoder.default(cast_in_data)
            if cls._content_type_is_json(content_type):
                value = cls._serialize_json(cast_in_data)
                return cls._to_dict(cls.name, value)
            raise NotImplementedError('Serialization of {} has not yet been implemented'.format(content_type))


class QueryParameter(ParameterBase, StyleFormSerializer):
    name: str
    required: bool = False
    in_type: ParameterInType = ParameterInType.QUERY
    style: ParameterStyle = ParameterStyle.FORM
    explode: typing.Optional[bool] = None
    allow_reserved: typing.Optional[bool] = None
    schema: typing.Optional[typing.Type[schemas.Schema]] = None
    content: typing.Optional[typing.Dict[str, typing.Type[schemas.Schema]]] = None

    @classmethod
    def __serialize_space_delimited(
        cls,
        in_data: typing.Union[None, int, float, str, bool, dict, list],
        prefix_separator_iterator: typing.Optional[PrefixSeparatorIterator]
    ) -> typing.Dict[str, str]:
        if prefix_separator_iterator is None:
            prefix_separator_iterator = cls.get_prefix_separator_iterator()
        value = cls._ref6570_expansion(
            variable_name=cls.name,
            in_data=in_data,
            explode=cls.explode,
            percent_encode=True,
            prefix_separator_iterator=prefix_separator_iterator
        )
        return cls._to_dict(cls.name, value)

    @classmethod
    def __serialize_pipe_delimited(
        cls,
        in_data: typing.Union[None, int, float, str, bool, dict, list],
        prefix_separator_iterator: typing.Optional[PrefixSeparatorIterator]
    ) -> typing.Dict[str, str]:
        if prefix_separator_iterator is None:
            prefix_separator_iterator = cls.get_prefix_separator_iterator()
        value = cls._ref6570_expansion(
            variable_name=cls.name,
            in_data=in_data,
            explode=cls.explode,
            percent_encode=True,
            prefix_separator_iterator=prefix_separator_iterator
        )
        return cls._to_dict(cls.name, value)

    @classmethod
    def __serialize_form(
        cls,
        in_data: typing.Union[None, int, float, str, bool, dict, list],
        prefix_separator_iterator: typing.Optional[PrefixSeparatorIterator]
    ) -> typing.Dict[str, str]:
        if prefix_separator_iterator is None:
            prefix_separator_iterator = cls.get_prefix_separator_iterator()
        value = cls._serialize_form(
            in_data,
            name=cls.name,
            explode=cls.explode,
            percent_encode=True,
            prefix_separator_iterator=prefix_separator_iterator
        )
        return cls._to_dict(cls.name, value)

    @classmethod
    def get_prefix_separator_iterator(cls) -> typing.Optional[PrefixSeparatorIterator]:
        if cls.style is ParameterStyle.FORM:
            return PrefixSeparatorIterator('?', '&')
        elif cls.style is ParameterStyle.SPACE_DELIMITED:
            return PrefixSeparatorIterator('', '%20')
        elif cls.style is ParameterStyle.PIPE_DELIMITED:
            return PrefixSeparatorIterator('', '|')

    @classmethod
    def serialize(
        cls,
        in_data: typing.Union[
            schemas.Schema, decimal.Decimal, int, float, str, datetime.date, datetime.datetime, None, bool, list, tuple, dict, frozendict.frozendict],
        prefix_separator_iterator: typing.Optional[PrefixSeparatorIterator] = None
    ) -> typing.Dict[str, str]:
        if cls.schema:
            cast_in_data = cls.schema(in_data)
            cast_in_data = cls._json_encoder.default(cast_in_data)
            """
            form -> query
                query:
                    - GET/HEAD/DELETE: could use fields
                    - PUT/POST: must use urlencode to send parameters
                    returns fields: tuple
            spaceDelimited -> query
                returns fields
            pipeDelimited -> query
                returns fields
            deepObject -> query, https://github.com/OAI/OpenAPI-Specification/issues/1706
                returns fields
            """
            if cls.style:
                # TODO update query ones to omit setting values when [] {} or None is input
                if cls.style is ParameterStyle.FORM:
                    return cls.__serialize_form(cast_in_data, prefix_separator_iterator)
                elif cls.style is ParameterStyle.SPACE_DELIMITED:
                    return cls.__serialize_space_delimited(cast_in_data, prefix_separator_iterator)
                elif cls.style is ParameterStyle.PIPE_DELIMITED:
                    return cls.__serialize_pipe_delimited(cast_in_data, prefix_separator_iterator)
        # cls.content will be length one
        if prefix_separator_iterator is None:
            prefix_separator_iterator = cls.get_prefix_separator_iterator()
        for content_type, schema in cls.content.items():
            cast_in_data = schema(in_data)
            cast_in_data = cls._json_encoder.default(cast_in_data)
            if cls._content_type_is_json(content_type):
                value = cls._serialize_json(cast_in_data, eliminate_whitespace=True)
                return cls._to_dict(
                    cls.name,
                    next(prefix_separator_iterator) + cls.name + '=' + parse.quote(value)
                )
            raise NotImplementedError('Serialization of {} has not yet been implemented'.format(content_type))


class CookieParameter(ParameterBase, StyleFormSerializer):
    name: str
    required: bool = False
    style: ParameterStyle = ParameterStyle.FORM
    in_type: ParameterInType = ParameterInType.COOKIE
    explode: typing.Optional[bool] = None
    allow_reserved: typing.Optional[bool] = None
    schema: typing.Optional[typing.Type[schemas.Schema]] = None
    content: typing.Optional[typing.Dict[str, typing.Type[schemas.Schema]]] = None

    @classmethod
    def serialize(
        cls,
        in_data: typing.Union[
            schemas.Schema, decimal.Decimal, int, float, str, datetime.date, datetime.datetime, None, bool, list, tuple, dict, frozendict.frozendict]
    ) -> typing.Dict[str, str]:
        if cls.schema:
            cast_in_data = cls.schema(in_data)
            cast_in_data = cls._json_encoder.default(cast_in_data)
            """
            form -> cookie
                returns fields: tuple
            """
            if cls.style:
                """
                TODO add escaping of comma, space, equals
                or turn encoding on
                """
                value = cls._serialize_form(
                    cast_in_data,
                    explode=cls.explode,
                    name=cls.name,
                    percent_encode=False,
                    prefix_separator_iterator=PrefixSeparatorIterator('', '&')
                )
                return cls._to_dict(cls.name, value)
        # cls.content will be length one
        for content_type, schema in cls.content.items():
            cast_in_data = schema(in_data)
            cast_in_data = cls._json_encoder.default(cast_in_data)
            if cls._content_type_is_json(content_type):
                value = cls._serialize_json(cast_in_data)
                return cls._to_dict(cls.name, value)
            raise NotImplementedError('Serialization of {} has not yet been implemented'.format(content_type))


class HeaderParameterWithoutName(ParameterBase, StyleSimpleSerializer):
    required: bool = False
    style: ParameterStyle = ParameterStyle.SIMPLE
    in_type: ParameterInType = ParameterInType.HEADER
    explode: bool = False
    allow_reserved: typing.Optional[bool] = None
    schema: typing.Optional[typing.Type[schemas.Schema]] = None
    content: typing.Optional[typing.Dict[str, typing.Type[schemas.Schema]]] = None

    @staticmethod
    def __to_headers(in_data: typing.Tuple[typing.Tuple[str, str], ...]) -> _collections.HTTPHeaderDict:
        data = tuple(t for t in in_data if t)
        headers = _collections.HTTPHeaderDict()
        if not data:
            return headers
        headers.extend(data)
        return headers

    @classmethod
    def serialize(
        cls,
        in_data: typing.Union[
            schemas.Schema, decimal.Decimal, int, float, str, datetime.date, datetime.datetime, None, bool, list, tuple, dict, frozendict.frozendict],
        name: str
    ) -> _collections.HTTPHeaderDict:
        if cls.schema:
            cast_in_data = cls.schema(in_data)
            cast_in_data = cls._json_encoder.default(cast_in_data)
            """
            simple -> header
                headers: PoolManager needs a mapping, tuple is close
                    returns headers: dict
            """
            if cls.style:
                value = cls._serialize_simple(cast_in_data, name, cls.explode, False)
                return cls.__to_headers(((name, value),))
        # cls.content will be length one
        for content_type, schema in cls.content.items():
            cast_in_data = schema(in_data)
            cast_in_data = cls._json_encoder.default(cast_in_data)
            if cls._content_type_is_json(content_type):
                value = cls._serialize_json(cast_in_data)
                return cls.__to_headers(((name, value),))
            raise NotImplementedError('Serialization of {} has not yet been implemented'.format(content_type))

    @classmethod
    def deserialize(
        cls,
        in_data: str,
        name: str
    ) -> schemas.Schema:
        if cls.schema:
            """
            simple -> header
                headers: PoolManager needs a mapping, tuple is close
                    returns headers: dict
            """
            if cls.style:
                extracted_data = cls._deserialize_simple(in_data, name, cls.explode, False)
                return schema.from_openapi_data_oapg(extracted_data)
        # cls.content will be length one
        for content_type, schema in cls.content.items():
            if cls._content_type_is_json(content_type):
                cast_in_data = json.loads(in_data)
                return schema.from_openapi_data_oapg(cast_in_data)
            raise NotImplementedError('Deserialization of {} has not yet been implemented'.format(content_type))


class HeaderParameter(HeaderParameterWithoutName):
    name: str

    @classmethod
    def serialize(
        cls,
        in_data: typing.Union[
            schemas.Schema, decimal.Decimal, int, float, str, datetime.date, datetime.datetime, None, bool, list, tuple, dict, frozendict.frozendict]
    ) -> _collections.HTTPHeaderDict:
        return super().serialize(
            in_data,
            cls.name
        )


class Encoding:
    def __init__(
        self,
        content_type: str,
        headers: typing.Optional[typing.Dict[str, HeaderParameter]] = None,
        style: typing.Optional[ParameterStyle] = None,
        explode: bool = False,
        allow_reserved: bool = False,
    ):
        self.content_type = content_type
        self.headers = headers
        self.style = style
        self.explode = explode
        self.allow_reserved = allow_reserved


@dataclasses.dataclass
class MediaType:
    """
    Used to store request and response body schema information
    encoding:
        A map between a property name and its encoding information.
        The key, being the property name, MUST exist in the schema as a property.
        The encoding object SHALL only apply to requestBody objects when the media type is
        multipart or application/x-www-form-urlencoded.
    """
    schema: typing.Optional[typing.Type[schemas.Schema]] = None
    encoding: typing.Optional[typing.Dict[str, Encoding]] = None


@dataclasses.dataclass
class ApiResponse:
    response: urllib3.HTTPResponse
    body: typing.Union[schemas.Unset, schemas.Schema] = schemas.unset
    headers: typing.Union[schemas.Unset, typing.Dict[str, schemas.Schema]] = schemas.unset

    def __init__(
        self,
        response: urllib3.HTTPResponse,
        body: typing.Union[schemas.Unset, schemas.Schema] = schemas.unset,
        headers: typing.Union[schemas.Unset, typing.Dict[str, schemas.Schema]] = schemas.unset
    ):
        """
        pycharm needs this to prevent 'Unexpected argument' warnings
        """
        self.response = response
        self.body = body
        self.headers = headers


@dataclasses.dataclass
class ApiResponseWithoutDeserialization(ApiResponse):
    response: urllib3.HTTPResponse
    body: typing.Union[schemas.Unset, schemas.Schema] = schemas.unset
    headers: typing.Union[schemas.Unset, typing.Dict[str, schemas.Schema]] = schemas.unset


class TypedDictInputVerifier:
    @staticmethod
    def _verify_typed_dict_inputs_oapg(cls: typing.Type[typing_extensions.TypedDict], data: typing.Dict[str, typing.Any]):
        """
        Ensures that:
        - required keys are present
        - additional properties are not input
        - value stored under required keys do not have the value schemas.unset
        Note: detailed value checking is done in schema classes
        """
        missing_required_keys = []
        required_keys_with_unset_values = []
        for required_key in cls.__required_keys__:
            if required_key not in data:
                missing_required_keys.append(required_key)
                continue
            value = data[required_key]
            if value is schemas.unset:
                required_keys_with_unset_values.append(required_key)
        if missing_required_keys:
            raise exceptions.ApiTypeError(
                '{} missing {} required arguments: {}'.format(
                    cls.__name__, len(missing_required_keys), missing_required_keys
                 )
             )
        if required_keys_with_unset_values:
            raise exceptions.ApiValueError(
                '{} contains invalid unset values for {} required keys: {}'.format(
                    cls.__name__, len(required_keys_with_unset_values), required_keys_with_unset_values
                )
            )

        disallowed_additional_keys = []
        for key in data:
            if key in cls.__required_keys__ or key in cls.__optional_keys__:
                continue
            disallowed_additional_keys.append(key)
        if disallowed_additional_keys:
            raise exceptions.ApiTypeError(
                '{} got {} unexpected keyword arguments: {}'.format(
                    cls.__name__, len(disallowed_additional_keys), disallowed_additional_keys
                )
            )


T = typing.TypeVar("T")


@dataclasses.dataclass
class OpenApiResponse(JSONDetector, TypedDictInputVerifier, typing.Generic[T]):
    __filename_content_disposition_pattern = re.compile('filename="(.+?)"')
    response_cls: typing.Type[T]
    content: typing.Optional[typing.Dict[str, MediaType]] = None
    headers: typing.Optional[typing.Dict[str, HeaderParameterWithoutName]] = None

    @staticmethod
    def __deserialize_json(response: urllib3.HTTPResponse) -> typing.Any:
        # python must be >= 3.9 so we can pass in bytes into json.loads
        return json.loads(response.data)

    @staticmethod
    def __file_name_from_response_url(response_url: typing.Optional[str]) -> typing.Optional[str]:
        if response_url is None:
            return None
        url_path = parse.urlparse(response_url).path
        if url_path:
            path_basename = os.path.basename(url_path)
            if path_basename:
                _filename, ext = os.path.splitext(path_basename)
                if ext:
                    return path_basename
        return None

    @classmethod
    def __file_name_from_content_disposition(cls, content_disposition: typing.Optional[str]) -> typing.Optional[str]:
        if content_disposition is None:
            return None
        match = cls.__filename_content_disposition_pattern.search(content_disposition)
        if not match:
            return None
        return match.group(1)

    @classmethod
    def __deserialize_application_octet_stream(
        cls, response: urllib3.HTTPResponse
    ) -> typing.Union[bytes, io.BufferedReader]:
        """
        urllib3 use cases:
        1. when preload_content=True (stream=False) then supports_chunked_reads is False and bytes are returned
        2. when preload_content=False (stream=True) then supports_chunked_reads is True and
            a file will be written and returned
        """
        if response.supports_chunked_reads():
            file_name = (
                cls.__file_name_from_content_disposition(response.headers.get('content-disposition'))
                or cls.__file_name_from_response_url(response.geturl())
            )

            if file_name is None:
                _fd, path = tempfile.mkstemp()
            else:
                path = os.path.join(tempfile.gettempdir(), file_name)

            with open(path, 'wb') as new_file:
                chunk_size = 1024
                while True:
                    data = response.read(chunk_size)
                    if not data:
                        break
                    new_file.write(data)
            # release_conn is needed for streaming connections only
            response.release_conn()
            new_file = open(path, 'rb')
            return new_file
        else:
            return response.data

    @staticmethod
    def __deserialize_multipart_form_data(
        response: urllib3.HTTPResponse
    ) -> typing.Dict[str, typing.Any]:
        msg = email.message_from_bytes(response.data)
        return {
            part.get_param("name", header="Content-Disposition"): part.get_payload(
                decode=True
            ).decode(part.get_content_charset())
            if part.get_content_charset()
            else part.get_payload()
            for part in msg.get_payload()
        }

    @classmethod
    def deserialize(cls, response: urllib3.HTTPResponse, configuration: configuration_module.Configuration) -> T:
        content_type = response.getheader('content-type')
        deserialized_body = schemas.unset
        streamed = response.supports_chunked_reads()

        deserialized_headers = schemas.unset
        if cls.headers is not None:
            cls._verify_typed_dict_inputs_oapg(cls.response_cls.headers, response.headers)
            deserialized_headers = {}
            for header_name, header_param in self.headers.items():
                header_value = response.getheader(header_name)
                if header_value is None:
                    continue
                header_value = header_param.deserialize(header_value, header_name)
                deserialized_headers[header_name] = header_value

        if cls.content is not None:
            if content_type not in cls.content:
                raise exceptions.ApiValueError(
                    f"Invalid content_type returned. Content_type='{content_type}' was returned "
                    f"when only {str(set(cls.content))} are defined for status_code={str(response.status)}"
                )
            body_schema = cls.content[content_type].schema
            if body_schema is None:
                # some specs do not define response content media type schemas
                return cls.response_cls(
                    response=response,
                    headers=deserialized_headers,
                    body=schemas.unset
                )

            if cls._content_type_is_json(content_type):
                body_data = cls.__deserialize_json(response)
            elif content_type == 'application/octet-stream':
                body_data = cls.__deserialize_application_octet_stream(response)
            elif content_type.startswith('multipart/form-data'):
                body_data = cls.__deserialize_multipart_form_data(response)
                content_type = 'multipart/form-data'
            else:
                raise NotImplementedError('Deserialization of {} has not yet been implemented'.format(content_type))
            deserialized_body = body_schema.from_openapi_data_oapg(
                body_data, _configuration=configuration)
        elif streamed:
            response.release_conn()

        return cls.response_cls(
            response=response,
            headers=deserialized_headers,
            body=deserialized_body
        )


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_module.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.
    """

    _pool = None

    def __init__(
        self,
        configuration: typing.Optional[configuration_module.Configuration] = None,
        header_name: typing.Optional[str] = None,
        header_value: typing.Optional[str] = None,
        cookie: typing.Optional[str] = None,
        pool_threads: int = 1
    ):
        if configuration is None:
            configuration = configuration_module.Configuration()
        self.configuration = configuration
        self.pool_threads = pool_threads

        self.rest_client = rest.RESTClientObject(configuration)
        self.default_headers = _collections.HTTPHeaderDict()
        if header_name is not None:
            self.default_headers[header_name] = header_value
        self.cookie = cookie
        # Set default User-Agent.
        self.user_agent = '{{#if httpUserAgent}}{{{httpUserAgent}}}{{/if}}{{#unless httpUserAgent}}OpenAPI-Generator/{{{packageVersion}}}/python{{/unless}}'

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

    def close(self):
        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 = 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

    {{#if tornado}}
    @tornado.gen.coroutine
    {{/if}}
    {{#if asyncio}}async {{/if}}def __call_api(
        self,
        resource_path: str,
        method: str,
        headers: typing.Optional[_collections.HTTPHeaderDict] = None,
        body: typing.Optional[typing.Union[str, bytes]] = None,
        fields: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
        auth_settings: typing.Optional[typing.List[str]] = None,
        stream: bool = False,
        timeout: typing.Optional[typing.Union[int, typing.Tuple]] = None,
        host: typing.Optional[str] = None,
    ) -> urllib3.HTTPResponse:

        # header parameters
        used_headers = _collections.HTTPHeaderDict(self.default_headers)
        if self.cookie:
            headers['Cookie'] = self.cookie

        # auth setting
        self.update_params_for_auth(used_headers,
                                    auth_settings, resource_path, method, body)

        # must happen after cookie setting and auth setting in case user is overriding those
        if headers:
            used_headers.update(headers)

        # request url
        if host is None:
            url = self.configuration.host + resource_path
        else:
            # use server/host defined in path or operation instead
            url = host + resource_path

        # perform request and return response
        response = {{#if asyncio}}await {{/if}}{{#if tornado}}yield {{/if}}self.request(
            method,
            url,
            headers=used_headers,
            fields=fields,
            body=body,
            stream=stream,
            timeout=timeout,
        )
        return response

    def call_api(
        self,
        resource_path: str,
        method: str,
        headers: typing.Optional[_collections.HTTPHeaderDict] = None,
        body: typing.Optional[typing.Union[str, bytes]] = None,
        fields: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
        auth_settings: typing.Optional[typing.List[str]] = None,
        async_req: typing.Optional[bool] = None,
        stream: bool = False,
        timeout: typing.Optional[typing.Union[int, typing.Tuple]] = None,
        host: typing.Optional[str] = None,
    ) -> urllib3.HTTPResponse:
        """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 headers: Header parameters to be
            placed in the request header.
        :param body: Request body.
        :param fields: Request post form parameters,
            for `application/x-www-form-urlencoded`, `multipart/form-data`.
        :param auth_settings: Auth Settings names for the request.
        :param async_req: execute request asynchronously
        :type async_req: bool, optional TODO remove, unused
        :param stream: if True, the urllib3.HTTPResponse object will
                                 be returned without reading/decoding response
                                 data. Also when True, if the openapi spec describes a file download,
                                 the data will be written to a local filesystme file and the schemas.BinarySchema
                                 instance will also inherit from FileSchema and schemas.FileIO
                                 Default is False.
        :type stream: bool, optional
        :param 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 host: api endpoint host
        :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 self.__call_api(
                resource_path,
                method,
                headers,
                body,
                fields,
                auth_settings,
                stream,
                timeout,
                host,
            )

        return self.pool.apply_async(
            self.__call_api,
            (
                resource_path,
                method,
                headers,
                body,
                json,
                fields,
                auth_settings,
                stream,
                timeout,
                host,
            )
        )

    def request(
        self,
        method: str,
        url: str,
        headers: typing.Optional[_collections.HTTPHeaderDict] = None,
        fields: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
        body: typing.Optional[typing.Union[str, bytes]] = None,
        stream: bool = False,
        timeout: typing.Optional[typing.Union[int, typing.Tuple]] = None,
    ) -> urllib3.HTTPResponse:
        """Makes the HTTP request using RESTClient."""
        if method == "GET":
            return self.rest_client.GET(url,
                                        stream=stream,
                                        timeout=timeout,
                                        headers=headers)
        elif method == "HEAD":
            return self.rest_client.HEAD(url,
                                         stream=stream,
                                         timeout=timeout,
                                         headers=headers)
        elif method == "OPTIONS":
            return self.rest_client.OPTIONS(url,
                                            headers=headers,
                                            fields=fields,
                                            stream=stream,
                                            timeout=timeout,
                                            body=body)
        elif method == "POST":
            return self.rest_client.POST(url,
                                         headers=headers,
                                         fields=fields,
                                         stream=stream,
                                         timeout=timeout,
                                         body=body)
        elif method == "PUT":
            return self.rest_client.PUT(url,
                                        headers=headers,
                                        fields=fields,
                                        stream=stream,
                                        timeout=timeout,
                                        body=body)
        elif method == "PATCH":
            return self.rest_client.PATCH(url,
                                          headers=headers,
                                          fields=fields,
                                          stream=stream,
                                          timeout=timeout,
                                          body=body)
        elif method == "DELETE":
            return self.rest_client.DELETE(url,
                                           headers=headers,
                                           stream=stream,
                                           timeout=timeout,
                                           body=body)
        else:
            raise exceptions.ApiValueError(
                "http method must be `GET`, `HEAD`, `OPTIONS`,"
                " `POST`, `PATCH`, `PUT` or `DELETE`."
            )

    def update_params_for_auth(self, headers, auth_settings,
                               resource_path, method, body):
        """Updates header and query params based on authentication setting.

        :param headers: Header parameters dict to be updated.
        :param auth_settings: Authentication setting identifiers list.
        :param resource_path: A string representation of the HTTP request resource path.
        :param method: A string representation of the HTTP request method.
        :param body: A object representing the body of the HTTP request.
            The object type is the return value of _encoder.default().
        """
        if not auth_settings:
            return

        for auth in auth_settings:
            auth_setting = self.configuration.auth_settings().get(auth)
            if not auth_setting:
                continue
            if auth_setting['in'] == 'cookie':
                headers.add('Cookie', auth_setting['value'])
            elif auth_setting['in'] == 'header':
                if auth_setting['type'] != 'http-signature':
                    headers.add(auth_setting['key'], auth_setting['value'])
{{#if hasHttpSignatureMethods}}
                else:
                    # The HTTP signature scheme requires multiple HTTP headers
                    # that are calculated dynamically.
                    signing_info = self.configuration.signing_info
                    querys = tuple()
                    auth_headers = signing_info.get_http_signature_headers(
                                        resource_path, method, headers, body, querys)
                    for key, value in auth_headers.items():
                        headers.add(key, value)
{{/if}}
            elif auth_setting['in'] == 'query':
                """ TODO implement auth in query
                need to pass in prefix_separator_iterator
                and need to output resource_path with query params added
                """
                raise exceptions.ApiValueError("Auth in query not yet implemented")
            else:
                raise exceptions.ApiValueError(
                    'Authentication token must be in `query` or `header`'
                )


class Api(TypedDictInputVerifier):
    """NOTE: This class is auto generated by OpenAPI Generator
    Ref: https://openapi-generator.tech

    Do not edit the class manually.
    """

    def __init__(self, api_client: typing.Optional[ApiClient] = None):
        if api_client is None:
            api_client = ApiClient()
        self.api_client = api_client

    def _get_host_oapg(
        self,
        operation_id: str,
        servers: typing.Tuple[typing.Dict[str, str], ...] = tuple(),
        host_index: typing.Optional[int] = None
    ) -> typing.Optional[str]:
        configuration = self.api_client.configuration
        try:
            if host_index is None:
                index = configuration.server_operation_index.get(
                    operation_id, configuration.server_index
                )
            else:
                index = host_index
            server_variables = configuration.server_operation_variables.get(
                operation_id, configuration.server_variables
            )
            host = configuration.get_host_from_settings(
                index, variables=server_variables, servers=servers
            )
        except IndexError:
            if servers:
                raise exceptions.ApiValueError(
                    "Invalid host index. Must be 0 <= index < %s" %
                    len(servers)
                )
            host = None
        return host


class SerializedRequestBody(typing_extensions.TypedDict, total=False):
    body: typing.Union[str, bytes]
    fields: typing.Tuple[typing.Union[RequestField, typing.Tuple[str, str]], ...]


class RequestBody(StyleFormSerializer, JSONDetector):
    """
    A request body parameter
    content: content_type to MediaType schemas.Schema info
    """
    __json_encoder = JSONEncoder()
    content: typing.Dict[str, MediaType]
    required: bool = False

    @classmethod
    def __serialize_json(
        cls,
        in_data: typing.Any
    ) -> typing.Dict[str, bytes]:
        in_data = cls.__json_encoder.default(in_data)
        json_str = json.dumps(in_data, separators=(",", ":"), ensure_ascii=False).encode(
            "utf-8"
        )
        return dict(body=json_str)

    @staticmethod
    def __serialize_text_plain(in_data: typing.Any) -> typing.Dict[str, str]:
        if isinstance(in_data, frozendict.frozendict):
            raise ValueError('Unable to serialize type frozendict.frozendict to text/plain')
        elif isinstance(in_data, tuple):
            raise ValueError('Unable to serialize type tuple to text/plain')
        elif isinstance(in_data, schemas.NoneClass):
            raise ValueError('Unable to serialize type NoneClass to text/plain')
        elif isinstance(in_data, schemas.BoolClass):
            raise ValueError('Unable to serialize type BoolClass to text/plain')
        return dict(body=str(in_data))

    @classmethod
    def __multipart_json_item(cls, key: str, value: schemas.Schema) -> RequestField:
        json_value = cls.__json_encoder.default(value)
        request_field = RequestField(name=key, data=json.dumps(json_value))
        request_field.make_multipart(content_type='application/json')
        return request_field

    @classmethod
    def __multipart_form_item(cls, key: str, value: schemas.Schema) -> RequestField:
        if isinstance(value, str):
            request_field = RequestField(name=key, data=str(value))
            request_field.make_multipart(content_type='text/plain')
        elif isinstance(value, bytes):
            request_field = RequestField(name=key, data=value)
            request_field.make_multipart(content_type='application/octet-stream')
        elif isinstance(value, schemas.FileIO):
            # TODO use content.encoding to limit allowed content types if they are present
            request_field = RequestField.from_tuples(key, (os.path.basename(value.name), value.read()))
            value.close()
        else:
            request_field = cls.__multipart_json_item(key=key, value=value)
        return request_field

    @classmethod
    def __serialize_multipart_form_data(
        cls, in_data: schemas.Schema
    ) -> typing.Dict[str, typing.Tuple[RequestField, ...]]:
        if not isinstance(in_data, frozendict.frozendict):
            raise ValueError(f'Unable to serialize {in_data} to multipart/form-data because it is not a dict of data')
        """
        In a multipart/form-data request body, each schema property, or each element of a schema array property,
        takes a section in the payload with an internal header as defined by RFC7578. The serialization strategy
        for each property of a multipart/form-data request body can be specified in an associated Encoding Object.

        When passing in multipart types, boundaries MAY be used to separate sections of the content being
        transferred – thus, the following default Content-Types are defined for multipart:

        If the (object) property is a primitive, or an array of primitive values, the default Content-Type is text/plain
        If the property is complex, or an array of complex values, the default Content-Type is application/json
            Question: how is the array of primitives encoded?
        If the property is a type: string with a contentEncoding, the default Content-Type is application/octet-stream
        """
        fields = []
        for key, value in in_data.items():
            if isinstance(value, tuple):
                if value:
                    # values use explode = True, so the code makes a RequestField for each item with name=key
                    for item in value:
                        request_field = cls.__multipart_form_item(key=key, value=item)
                        fields.append(request_field)
                else:
                    # send an empty array as json because exploding will not send it
                    request_field = cls.__multipart_json_item(key=key, value=value)
                    fields.append(request_field)
            else:
                request_field = cls.__multipart_form_item(key=key, value=value)
                fields.append(request_field)

        return dict(fields=tuple(fields))

    @staticmethod
    def __serialize_application_octet_stream(in_data: schemas.BinarySchema) -> typing.Dict[str, bytes]:
        if isinstance(in_data, bytes):
            return dict(body=in_data)
        # schemas.FileIO type
        result = dict(body=in_data.read())
        in_data.close()
        return result

    @classmethod
    def __serialize_application_x_www_form_data(
        cls, in_data: typing.Any
    ) -> SerializedRequestBody:
        """
        POST submission of form data in body
        """
        if not isinstance(in_data, frozendict.frozendict):
            raise ValueError(
                f'Unable to serialize {in_data} to application/x-www-form-urlencoded because it is not a dict of data')
        cast_in_data = cls.__json_encoder.default(in_data)
        value = cls._serialize_form(cast_in_data, name='', explode=True, percent_encode=True)
        return dict(body=value)

    @classmethod
    def serialize(
        cls, in_data: typing.Any, content_type: str
    ) -> SerializedRequestBody:
        """
        If a str is returned then the result will be assigned to data when making the request
        If a tuple is returned then the result will be used as fields input in encode_multipart_formdata
        Return a tuple of

        The key of the return dict is
        - body for application/json
        - encode_multipart and fields for multipart/form-data
        """
        media_type = cls.content[content_type]
        if isinstance(in_data, media_type.schema):
            cast_in_data = in_data
        elif isinstance(in_data, (dict, frozendict.frozendict)) and in_data:
            cast_in_data = media_type.schema(**in_data)
        else:
            cast_in_data = media_type.schema(in_data)
        # TODO check for and use encoding if it exists
        # and content_type is multipart or application/x-www-form-urlencoded
        if cls._content_type_is_json(content_type):
            return cls.__serialize_json(cast_in_data)
        elif content_type == 'text/plain':
            return cls.__serialize_text_plain(cast_in_data)
        elif content_type == 'multipart/form-data':
            return cls.__serialize_multipart_form_data(cast_in_data)
        elif content_type == 'application/x-www-form-urlencoded':
            return cls.__serialize_application_x_www_form_data(cast_in_data)
        elif content_type == 'application/octet-stream':
            return cls.__serialize_application_octet_stream(cast_in_data)
        raise NotImplementedError('Serialization has not yet been implemented for {}'.format(content_type))