loafle/openapi-generator
3
0
Fork 0
forked from loafle/openapi-generator-original
Code Pull Requests Activity

[python] fix typing for API responses (#16802)

Browse Source 
* [python] remove _preload_content

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] remove _return_http_data_only

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] remove async_req

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] fix typing for API responses

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update samples

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] fix AttributeError

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] remove _preload_content

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] fix response_type

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] fix typo

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update simplify RESTClientObject

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update split call_api into 4 functions

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update samples

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update improve stream

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] remove kwargs

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update ApiResponse

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] add method for each return value

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update test

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update samples

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update docs

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] add constantParams

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] fix ImportError

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] fix SyntaxError in RESTResponse

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] add ApiResponse model_config

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] fix when isBinary is str

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update type

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update samples

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] add rest with pydantic v1

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update format

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update format and type

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] add test

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update type to pydantic strict type

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] remove leftover files

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] remove descriptions per field

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] add test

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] fix test

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] fix test

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] remove multiprocessing

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update samples

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] add blank line to docstring

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] update docstring

Signed-off-by: ふぁ <yuki@yuki0311.com>

* [python] remove unwanted imports in rest

Signed-off-by: ふぁ <yuki@yuki0311.com>

---------

Signed-off-by: ふぁ <yuki@yuki0311.com>
This commit is contained in:
ふぁ
2023-10-30 13:37:23 +09:00
committed by GitHub
parent 3422ef1e64
commit 8827da8012
238 changed files with 33185 additions and 17862 deletions
Download Patch File Download Diff File Expand all files Collapse all files

347
samples/client/echo_api/python/openapi_client/api/auth_api.py
View File

@@ -13,20 +13,21 @@
""" # noqa: E501
import re # noqa: F401
import io
import warnings
from pydantic import validate_call, ValidationError
from typing import Dict, List, Optional, Tuple
from pydantic import validate_call, Field, StrictFloat, StrictStr, StrictInt
from typing import Dict, List, Optional, Tuple, Union, Any
try:
from typing import Annotated
except ImportError:
from typing_extensions import Annotated
from openapi_client.api_client import ApiClient
from openapi_client.api_response import ApiResponse
from openapi_client.exceptions import ( # noqa: F401
ApiTypeError,
ApiValueError
)
from openapi_client.rest import RESTResponseType
class AuthApi:
@@ -41,143 +42,249 @@ class AuthApi:
api_client = ApiClient.get_default()
self.api_client = api_client
@validate_call
def test_auth_http_basic(
self,
**kwargs,
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[
Annotated[StrictFloat, Field(gt=0)],
Annotated[StrictFloat, Field(gt=0)]
]
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> str:
"""To test HTTP basic authentication # noqa: E501
"""To test HTTP basic authentication
To test HTTP basic authentication # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
To test HTTP basic authentication
>>> thread = api.test_auth_http_basic(async_req=True)
>>> result = thread.get()
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
: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: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: str
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the test_auth_http_basic_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
return self.test_auth_http_basic_with_http_info.raw_function(
**kwargs,
)
@validate_call
def test_auth_http_basic_with_http_info(
self,
**kwargs,
) -> ApiResponse:
"""To test HTTP basic authentication # noqa: E501
To test HTTP basic authentication # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.test_auth_http_basic_with_http_info(async_req=True)
>>> result = thread.get()
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _preload_content: if False, the ApiResponse.data will
be set to none and raw_data will store the
HTTP response body without reading/decoding.
Default is True.
:type _preload_content: bool, optional
:param _return_http_data_only: response data instead of ApiResponse
object with status code, headers, etc
:type _return_http_data_only: bool, optional
: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.
:type _request_timeout: int, tuple(int, int), optional
: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.
request; this effectively ignores the
authentication in the spec for a single request.
:type _request_auth: dict, optional
:type _content_type: string, optional: force content-type for the request
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: tuple(str, status_code(int), headers(HTTPHeaderDict))
"""
""" # noqa: E501
_params = locals()
_param = self._test_auth_http_basic_serialize(
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index
)
_all_params = [
]
_all_params.extend(
_response_types_map: Dict[str, Optional[str]] = {
'200': "str"
}
response_data = self.api_client.call_api(
*_param,
_request_timeout=_request_timeout
)
response_data.read()
return self.api_client.response_deserialize(
response_data=response_data,
response_types_map=_response_types_map,
).data
@validate_call
def test_auth_http_basic_with_http_info(
self,
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[
Annotated[StrictFloat, Field(gt=0)],
Annotated[StrictFloat, Field(gt=0)]
]
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> ApiResponse[str]:
"""To test HTTP basic authentication
To test HTTP basic authentication
: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.
:type _request_timeout: int, tuple(int, int), optional
: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.
:type _request_auth: dict, optional
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
""" # noqa: E501
_param = self._test_auth_http_basic_serialize(
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index
)
_response_types_map: Dict[str, Optional[str]] = {
'200': "str"
}
response_data = self.api_client.call_api(
*_param,
_request_timeout=_request_timeout
)
response_data.read()
return self.api_client.response_deserialize(
response_data=response_data,
response_types_map=_response_types_map,
)
@validate_call
def test_auth_http_basic_without_preload_content(
self,
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[
Annotated[StrictFloat, Field(gt=0)],
Annotated[StrictFloat, Field(gt=0)]
]
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> RESTResponseType:
"""To test HTTP basic authentication
To test HTTP basic authentication
: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.
:type _request_timeout: int, tuple(int, int), optional
: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.
:type _request_auth: dict, optional
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
""" # noqa: E501
_param = self._test_auth_http_basic_serialize(
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index
)
_response_types_map: Dict[str, Optional[str]] = {
'200': "str"
}
response_data = self.api_client.call_api(
*_param,
_request_timeout=_request_timeout
)
return response_data.response
def _test_auth_http_basic_serialize(
self,
_request_auth,
_content_type,
_headers,
_host_index,
) -> Tuple:
_host = None
_collection_formats: Dict[str, str] = {
}
_path_params: Dict[str, str] = {}
_query_params: List[Tuple[str, str]] = []
_header_params: Dict[str, Optional[str]] = _headers or {}
_form_params: List[Tuple[str, str]] = []
_files: Dict[str, str] = {}
_body_params: Optional[bytes] = None
# process the path parameters
# process the query parameters
# process the header parameters
# process the form parameters
# process the body parameter
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
'_request_auth',
'_content_type',
'_headers'
'text/plain'
]
)
# validate the arguments
for _key, _val in _params['kwargs'].items():
if _key not in _all_params:
raise ApiTypeError(
"Got an unexpected keyword argument '%s'"
" to method test_auth_http_basic" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats: Dict[str, str] = {}
# process the path parameters
_path_params: Dict[str, str] = {}
# process the query parameters
_query_params: List[Tuple[str, str]] = []
# process the header parameters
_header_params = dict(_params.get('_headers', {}))
# process the form parameters
_form_params: List[Tuple[str, str]] = []
_files: Dict[str, str] = {}
# process the body parameter
_body_params = None
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
['text/plain']) # noqa: E501
# authentication setting
_auth_settings: List[str] = ['http_auth'] # noqa: E501
_auth_settings: List[str] = [
'http_auth'
]
_response_types_map: Dict[str, Optional[str]] = {
'200': "str",
}
return self.api_client.call_api(
'/auth/http/basic', 'POST',
_path_params,
_query_params,
_header_params,
return self.api_client.param_serialize(
method='POST',
resource_path='/auth/http/basic',
path_params=_path_params,
query_params=_query_params,
header_params=_header_params,
body=_body_params,
post_params=_form_params,
files=_files,
response_types_map=_response_types_map,
auth_settings=_auth_settings,
async_req=_params.get('async_req'),
_return_http_data_only=_params.get('_return_http_data_only'), # noqa: E501
_preload_content=_params.get('_preload_content', True),
_request_timeout=_params.get('_request_timeout'),
collection_formats=_collection_formats,
_request_auth=_params.get('_request_auth'))
_host=_host,
_request_auth=_request_auth
)

2346
samples/client/echo_api/python/openapi_client/api/body_api.py
View File

File diff suppressed because it is too large Load Diff

803
samples/client/echo_api/python/openapi_client/api/form_api.py
View File

@@ -13,12 +13,16 @@
""" # noqa: E501
import re # noqa: F401
import io
import warnings
from pydantic import validate_call, ValidationError
from typing import Dict, List, Optional, Tuple
from pydantic import validate_call, Field, StrictFloat, StrictStr, StrictInt
from typing import Dict, List, Optional, Tuple, Union, Any
try:
from typing import Annotated
except ImportError:
from typing_extensions import Annotated
from pydantic import StrictBool, StrictInt, StrictStr
@@ -27,10 +31,7 @@ from typing import Optional
from openapi_client.api_client import ApiClient
from openapi_client.api_response import ApiResponse
from openapi_client.exceptions import ( # noqa: F401
ApiTypeError,
ApiValueError
)
from openapi_client.rest import RESTResponseType
class FormApi:
@@ -45,22 +46,29 @@ class FormApi:
api_client = ApiClient.get_default()
self.api_client = api_client
@validate_call
def test_form_integer_boolean_string(
self,
integer_form: Optional[StrictInt] = None,
boolean_form: Optional[StrictBool] = None,
string_form: Optional[StrictStr] = None,
**kwargs,
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[
Annotated[StrictFloat, Field(gt=0)],
Annotated[StrictFloat, Field(gt=0)]
]
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> str:
"""Test form parameter(s) # noqa: E501
"""Test form parameter(s)
Test form parameter(s) # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.test_form_integer_boolean_string(integer_form, boolean_form, string_form, async_req=True)
>>> result = thread.get()
Test form parameter(s)
:param integer_form:
:type integer_form: int
@@ -68,45 +76,75 @@ class FormApi:
:type boolean_form: bool
:param string_form:
:type string_form: str
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
: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_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.
:type _request_timeout: int, tuple(int, int), optional
: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.
:type _request_auth: dict, optional
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: str
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the test_form_integer_boolean_string_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
""" # noqa: E501
return self.test_form_integer_boolean_string_with_http_info.raw_function(
integer_form,
boolean_form,
string_form,
**kwargs,
_param = self._test_form_integer_boolean_string_serialize(
integer_form=integer_form,
boolean_form=boolean_form,
string_form=string_form,
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index
)
_response_types_map: Dict[str, Optional[str]] = {
'200': "str"
}
response_data = self.api_client.call_api(
*_param,
_request_timeout=_request_timeout
)
response_data.read()
return self.api_client.response_deserialize(
response_data=response_data,
response_types_map=_response_types_map,
).data
@validate_call
def test_form_integer_boolean_string_with_http_info(
self,
integer_form: Optional[StrictInt] = None,
boolean_form: Optional[StrictBool] = None,
string_form: Optional[StrictStr] = None,
**kwargs,
) -> ApiResponse:
"""Test form parameter(s) # noqa: E501
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[
Annotated[StrictFloat, Field(gt=0)],
Annotated[StrictFloat, Field(gt=0)]
]
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> ApiResponse[str]:
"""Test form parameter(s)
Test form parameter(s) # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.test_form_integer_boolean_string_with_http_info(integer_form, boolean_form, string_form, async_req=True)
>>> result = thread.get()
Test form parameter(s)
:param integer_form:
:type integer_form: int
@@ -114,117 +152,204 @@ class FormApi:
:type boolean_form: bool
:param string_form:
:type string_form: str
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _preload_content: if False, the ApiResponse.data will
be set to none and raw_data will store the
HTTP response body without reading/decoding.
Default is True.
:type _preload_content: bool, optional
:param _return_http_data_only: response data instead of ApiResponse
object with status code, headers, etc
:type _return_http_data_only: bool, optional
: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.
:type _request_timeout: int, tuple(int, int), optional
: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.
request; this effectively ignores the
authentication in the spec for a single request.
:type _request_auth: dict, optional
:type _content_type: string, optional: force content-type for the request
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: tuple(str, status_code(int), headers(HTTPHeaderDict))
"""
""" # noqa: E501
_params = locals()
_param = self._test_form_integer_boolean_string_serialize(
integer_form=integer_form,
boolean_form=boolean_form,
string_form=string_form,
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index
)
_all_params = [
'integer_form',
'boolean_form',
'string_form'
]
_all_params.extend(
_response_types_map: Dict[str, Optional[str]] = {
'200': "str"
}
response_data = self.api_client.call_api(
*_param,
_request_timeout=_request_timeout
)
response_data.read()
return self.api_client.response_deserialize(
response_data=response_data,
response_types_map=_response_types_map,
)
@validate_call
def test_form_integer_boolean_string_without_preload_content(
self,
integer_form: Optional[StrictInt] = None,
boolean_form: Optional[StrictBool] = None,
string_form: Optional[StrictStr] = None,
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[
Annotated[StrictFloat, Field(gt=0)],
Annotated[StrictFloat, Field(gt=0)]
]
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> RESTResponseType:
"""Test form parameter(s)
Test form parameter(s)
:param integer_form:
:type integer_form: int
:param boolean_form:
:type boolean_form: bool
:param string_form:
:type string_form: str
: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.
:type _request_timeout: int, tuple(int, int), optional
: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.
:type _request_auth: dict, optional
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
""" # noqa: E501
_param = self._test_form_integer_boolean_string_serialize(
integer_form=integer_form,
boolean_form=boolean_form,
string_form=string_form,
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index
)
_response_types_map: Dict[str, Optional[str]] = {
'200': "str"
}
response_data = self.api_client.call_api(
*_param,
_request_timeout=_request_timeout
)
return response_data.response
def _test_form_integer_boolean_string_serialize(
self,
integer_form,
boolean_form,
string_form,
_request_auth,
_content_type,
_headers,
_host_index,
) -> Tuple:
_host = None
_collection_formats: Dict[str, str] = {
}
_path_params: Dict[str, str] = {}
_query_params: List[Tuple[str, str]] = []
_header_params: Dict[str, Optional[str]] = _headers or {}
_form_params: List[Tuple[str, str]] = []
_files: Dict[str, str] = {}
_body_params: Optional[bytes] = None
# process the path parameters
# process the query parameters
# process the header parameters
# process the form parameters
if integer_form is not None:
_form_params.append(('integer_form', integer_form))
if boolean_form is not None:
_form_params.append(('boolean_form', boolean_form))
if string_form is not None:
_form_params.append(('string_form', string_form))
# process the body parameter
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
'_request_auth',
'_content_type',
'_headers'
'text/plain'
]
)
# validate the arguments
for _key, _val in _params['kwargs'].items():
if _key not in _all_params:
raise ApiTypeError(
"Got an unexpected keyword argument '%s'"
" to method test_form_integer_boolean_string" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats: Dict[str, str] = {}
# process the path parameters
_path_params: Dict[str, str] = {}
# process the query parameters
_query_params: List[Tuple[str, str]] = []
# process the header parameters
_header_params = dict(_params.get('_headers', {}))
# process the form parameters
_form_params: List[Tuple[str, str]] = []
_files: Dict[str, str] = {}
if _params['integer_form'] is not None:
_form_params.append(('integer_form', _params['integer_form']))
if _params['boolean_form'] is not None:
_form_params.append(('boolean_form', _params['boolean_form']))
if _params['string_form'] is not None:
_form_params.append(('string_form', _params['string_form']))
# process the body parameter
_body_params = None
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
['text/plain']) # noqa: E501
# set the HTTP header `Content-Type`
_content_types_list = _params.get('_content_type',
self.api_client.select_header_content_type(
['application/x-www-form-urlencoded']))
if _content_types_list:
_header_params['Content-Type'] = _content_types_list
if _content_type:
_header_params['Content-Type'] = _content_type
else:
_default_content_type = (
self.api_client.select_header_content_type(
[
'application/x-www-form-urlencoded'
]
)
)
if _default_content_type is not None:
_header_params['Content-Type'] = _default_content_type
# authentication setting
_auth_settings: List[str] = [] # noqa: E501
_auth_settings: List[str] = [
]
_response_types_map: Dict[str, Optional[str]] = {
'200': "str",
}
return self.api_client.call_api(
'/form/integer/boolean/string', 'POST',
_path_params,
_query_params,
_header_params,
return self.api_client.param_serialize(
method='POST',
resource_path='/form/integer/boolean/string',
path_params=_path_params,
query_params=_query_params,
header_params=_header_params,
body=_body_params,
post_params=_form_params,
files=_files,
response_types_map=_response_types_map,
auth_settings=_auth_settings,
async_req=_params.get('async_req'),
_return_http_data_only=_params.get('_return_http_data_only'), # noqa: E501
_preload_content=_params.get('_preload_content', True),
_request_timeout=_params.get('_request_timeout'),
collection_formats=_collection_formats,
_request_auth=_params.get('_request_auth'))
_host=_host,
_request_auth=_request_auth
)
@validate_call
def test_form_oneof(
@@ -235,16 +360,22 @@ class FormApi:
form4: Optional[StrictBool] = None,
id: Optional[StrictInt] = None,
name: Optional[StrictStr] = None,
**kwargs,
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[
Annotated[StrictFloat, Field(gt=0)],
Annotated[StrictFloat, Field(gt=0)]
]
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> str:
"""Test form parameter(s) for oneOf schema # noqa: E501
"""Test form parameter(s) for oneOf schema
Test form parameter(s) for oneOf schema # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.test_form_oneof(form1, form2, form3, form4, id, name, async_req=True)
>>> result = thread.get()
Test form parameter(s) for oneOf schema
:param form1:
:type form1: str
@@ -258,32 +389,56 @@ class FormApi:
:type id: int
:param name:
:type name: str
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
: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_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.
:type _request_timeout: int, tuple(int, int), optional
: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.
:type _request_auth: dict, optional
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: str
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the test_form_oneof_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
""" # noqa: E501
return self.test_form_oneof_with_http_info.raw_function(
form1,
form2,
form3,
form4,
id,
name,
**kwargs,
_param = self._test_form_oneof_serialize(
form1=form1,
form2=form2,
form3=form3,
form4=form4,
id=id,
name=name,
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index
)
_response_types_map: Dict[str, Optional[str]] = {
'200': "str"
}
response_data = self.api_client.call_api(
*_param,
_request_timeout=_request_timeout
)
response_data.read()
return self.api_client.response_deserialize(
response_data=response_data,
response_types_map=_response_types_map,
).data
@validate_call
def test_form_oneof_with_http_info(
self,
@@ -293,16 +448,22 @@ class FormApi:
form4: Optional[StrictBool] = None,
id: Optional[StrictInt] = None,
name: Optional[StrictStr] = None,
**kwargs,
) -> ApiResponse:
"""Test form parameter(s) for oneOf schema # noqa: E501
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[
Annotated[StrictFloat, Field(gt=0)],
Annotated[StrictFloat, Field(gt=0)]
]
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> ApiResponse[str]:
"""Test form parameter(s) for oneOf schema
Test form parameter(s) for oneOf schema # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.test_form_oneof_with_http_info(form1, form2, form3, form4, id, name, async_req=True)
>>> result = thread.get()
Test form parameter(s) for oneOf schema
:param form1:
:type form1: str
@@ -316,126 +477,224 @@ class FormApi:
:type id: int
:param name:
:type name: str
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _preload_content: if False, the ApiResponse.data will
be set to none and raw_data will store the
HTTP response body without reading/decoding.
Default is True.
:type _preload_content: bool, optional
:param _return_http_data_only: response data instead of ApiResponse
object with status code, headers, etc
:type _return_http_data_only: bool, optional
: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.
:type _request_timeout: int, tuple(int, int), optional
: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.
request; this effectively ignores the
authentication in the spec for a single request.
:type _request_auth: dict, optional
:type _content_type: string, optional: force content-type for the request
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: tuple(str, status_code(int), headers(HTTPHeaderDict))
"""
""" # noqa: E501
_params = locals()
_param = self._test_form_oneof_serialize(
form1=form1,
form2=form2,
form3=form3,
form4=form4,
id=id,
name=name,
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index
)
_all_params = [
'form1',
'form2',
'form3',
'form4',
'id',
'name'
]
_all_params.extend(
_response_types_map: Dict[str, Optional[str]] = {
'200': "str"
}
response_data = self.api_client.call_api(
*_param,
_request_timeout=_request_timeout
)
response_data.read()
return self.api_client.response_deserialize(
response_data=response_data,
response_types_map=_response_types_map,
)
@validate_call
def test_form_oneof_without_preload_content(
self,
form1: Optional[StrictStr] = None,
form2: Optional[StrictInt] = None,
form3: Optional[StrictStr] = None,
form4: Optional[StrictBool] = None,
id: Optional[StrictInt] = None,
name: Optional[StrictStr] = None,
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[
Annotated[StrictFloat, Field(gt=0)],
Annotated[StrictFloat, Field(gt=0)]
]
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> RESTResponseType:
"""Test form parameter(s) for oneOf schema
Test form parameter(s) for oneOf schema
:param form1:
:type form1: str
:param form2:
:type form2: int
:param form3:
:type form3: str
:param form4:
:type form4: bool
:param id:
:type id: int
:param name:
:type name: str
: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.
:type _request_timeout: int, tuple(int, int), optional
: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.
:type _request_auth: dict, optional
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
""" # noqa: E501
_param = self._test_form_oneof_serialize(
form1=form1,
form2=form2,
form3=form3,
form4=form4,
id=id,
name=name,
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index
)
_response_types_map: Dict[str, Optional[str]] = {
'200': "str"
}
response_data = self.api_client.call_api(
*_param,
_request_timeout=_request_timeout
)
return response_data.response
def _test_form_oneof_serialize(
self,
form1,
form2,
form3,
form4,
id,
name,
_request_auth,
_content_type,
_headers,
_host_index,
) -> Tuple:
_host = None
_collection_formats: Dict[str, str] = {
}
_path_params: Dict[str, str] = {}
_query_params: List[Tuple[str, str]] = []
_header_params: Dict[str, Optional[str]] = _headers or {}
_form_params: List[Tuple[str, str]] = []
_files: Dict[str, str] = {}
_body_params: Optional[bytes] = None
# process the path parameters
# process the query parameters
# process the header parameters
# process the form parameters
if form1 is not None:
_form_params.append(('form1', form1))
if form2 is not None:
_form_params.append(('form2', form2))
if form3 is not None:
_form_params.append(('form3', form3))
if form4 is not None:
_form_params.append(('form4', form4))
if id is not None:
_form_params.append(('id', id))
if name is not None:
_form_params.append(('name', name))
# process the body parameter
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
'_request_auth',
'_content_type',
'_headers'
'text/plain'
]
)
# validate the arguments
for _key, _val in _params['kwargs'].items():
if _key not in _all_params:
raise ApiTypeError(
"Got an unexpected keyword argument '%s'"
" to method test_form_oneof" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats: Dict[str, str] = {}
# process the path parameters
_path_params: Dict[str, str] = {}
# process the query parameters
_query_params: List[Tuple[str, str]] = []
# process the header parameters
_header_params = dict(_params.get('_headers', {}))
# process the form parameters
_form_params: List[Tuple[str, str]] = []
_files: Dict[str, str] = {}
if _params['form1'] is not None:
_form_params.append(('form1', _params['form1']))
if _params['form2'] is not None:
_form_params.append(('form2', _params['form2']))
if _params['form3'] is not None:
_form_params.append(('form3', _params['form3']))
if _params['form4'] is not None:
_form_params.append(('form4', _params['form4']))
if _params['id'] is not None:
_form_params.append(('id', _params['id']))
if _params['name'] is not None:
_form_params.append(('name', _params['name']))
# process the body parameter
_body_params = None
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
['text/plain']) # noqa: E501
# set the HTTP header `Content-Type`
_content_types_list = _params.get('_content_type',
self.api_client.select_header_content_type(
['application/x-www-form-urlencoded']))
if _content_types_list:
_header_params['Content-Type'] = _content_types_list
if _content_type:
_header_params['Content-Type'] = _content_type
else:
_default_content_type = (
self.api_client.select_header_content_type(
[
'application/x-www-form-urlencoded'
]
)
)
if _default_content_type is not None:
_header_params['Content-Type'] = _default_content_type
# authentication setting
_auth_settings: List[str] = [] # noqa: E501
_auth_settings: List[str] = [
]
_response_types_map: Dict[str, Optional[str]] = {
'200': "str",
}
return self.api_client.call_api(
'/form/oneof', 'POST',
_path_params,
_query_params,
_header_params,
return self.api_client.param_serialize(
method='POST',
resource_path='/form/oneof',
path_params=_path_params,
query_params=_query_params,
header_params=_header_params,
body=_body_params,
post_params=_form_params,
files=_files,
response_types_map=_response_types_map,
auth_settings=_auth_settings,
async_req=_params.get('async_req'),
_return_http_data_only=_params.get('_return_http_data_only'), # noqa: E501
_preload_content=_params.get('_preload_content', True),
_request_timeout=_params.get('_request_timeout'),
collection_formats=_collection_formats,
_request_auth=_params.get('_request_auth'))
_host=_host,
_request_auth=_request_auth
)

398
samples/client/echo_api/python/openapi_client/api/header_api.py
View File

@@ -13,12 +13,16 @@
""" # noqa: E501
import re # noqa: F401
import io
import warnings
from pydantic import validate_call, ValidationError
from typing import Dict, List, Optional, Tuple
from pydantic import validate_call, Field, StrictFloat, StrictStr, StrictInt
from typing import Dict, List, Optional, Tuple, Union, Any
try:
from typing import Annotated
except ImportError:
from typing_extensions import Annotated
from pydantic import StrictBool, StrictInt, StrictStr, field_validator
@@ -28,10 +32,7 @@ from openapi_client.models.string_enum_ref import StringEnumRef
from openapi_client.api_client import ApiClient
from openapi_client.api_response import ApiResponse
from openapi_client.exceptions import ( # noqa: F401
ApiTypeError,
ApiValueError
)
from openapi_client.rest import RESTResponseType
class HeaderApi:
@@ -46,6 +47,7 @@ class HeaderApi:
api_client = ApiClient.get_default()
self.api_client = api_client
@validate_call
def test_header_integer_boolean_string_enums(
self,
@@ -54,16 +56,22 @@ class HeaderApi:
string_header: Optional[StrictStr] = None,
enum_nonref_string_header: Optional[StrictStr] = None,
enum_ref_string_header: Optional[StringEnumRef] = None,
**kwargs,
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[
Annotated[StrictFloat, Field(gt=0)],
Annotated[StrictFloat, Field(gt=0)]
]
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> str:
"""Test header parameter(s) # noqa: E501
"""Test header parameter(s)
Test header parameter(s) # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.test_header_integer_boolean_string_enums(integer_header, boolean_header, string_header, enum_nonref_string_header, enum_ref_string_header, async_req=True)
>>> result = thread.get()
Test header parameter(s)
:param integer_header:
:type integer_header: int
@@ -75,31 +83,55 @@ class HeaderApi:
:type enum_nonref_string_header: str
:param enum_ref_string_header:
:type enum_ref_string_header: StringEnumRef
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
: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_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.
:type _request_timeout: int, tuple(int, int), optional
: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.
:type _request_auth: dict, optional
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: str
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the test_header_integer_boolean_string_enums_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
""" # noqa: E501
return self.test_header_integer_boolean_string_enums_with_http_info.raw_function(
integer_header,
boolean_header,
string_header,
enum_nonref_string_header,
enum_ref_string_header,
**kwargs,
_param = self._test_header_integer_boolean_string_enums_serialize(
integer_header=integer_header,
boolean_header=boolean_header,
string_header=string_header,
enum_nonref_string_header=enum_nonref_string_header,
enum_ref_string_header=enum_ref_string_header,
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index
)
_response_types_map: Dict[str, Optional[str]] = {
'200': "str"
}
response_data = self.api_client.call_api(
*_param,
_request_timeout=_request_timeout
)
response_data.read()
return self.api_client.response_deserialize(
response_data=response_data,
response_types_map=_response_types_map,
).data
@validate_call
def test_header_integer_boolean_string_enums_with_http_info(
self,
@@ -108,16 +140,22 @@ class HeaderApi:
string_header: Optional[StrictStr] = None,
enum_nonref_string_header: Optional[StrictStr] = None,
enum_ref_string_header: Optional[StringEnumRef] = None,
**kwargs,
) -> ApiResponse:
"""Test header parameter(s) # noqa: E501
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[
Annotated[StrictFloat, Field(gt=0)],
Annotated[StrictFloat, Field(gt=0)]
]
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> ApiResponse[str]:
"""Test header parameter(s)
Test header parameter(s) # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.test_header_integer_boolean_string_enums_with_http_info(integer_header, boolean_header, string_header, enum_nonref_string_header, enum_ref_string_header, async_req=True)
>>> result = thread.get()
Test header parameter(s)
:param integer_header:
:type integer_header: int
@@ -129,115 +167,203 @@ class HeaderApi:
:type enum_nonref_string_header: str
:param enum_ref_string_header:
:type enum_ref_string_header: StringEnumRef
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _preload_content: if False, the ApiResponse.data will
be set to none and raw_data will store the
HTTP response body without reading/decoding.
Default is True.
:type _preload_content: bool, optional
:param _return_http_data_only: response data instead of ApiResponse
object with status code, headers, etc
:type _return_http_data_only: bool, optional
: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.
:type _request_timeout: int, tuple(int, int), optional
: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.
request; this effectively ignores the
authentication in the spec for a single request.
:type _request_auth: dict, optional
:type _content_type: string, optional: force content-type for the request
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: tuple(str, status_code(int), headers(HTTPHeaderDict))
"""
""" # noqa: E501
_params = locals()
_param = self._test_header_integer_boolean_string_enums_serialize(
integer_header=integer_header,
boolean_header=boolean_header,
string_header=string_header,
enum_nonref_string_header=enum_nonref_string_header,
enum_ref_string_header=enum_ref_string_header,
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index
)
_all_params = [
'integer_header',
'boolean_header',
'string_header',
'enum_nonref_string_header',
'enum_ref_string_header'
]
_all_params.extend(
_response_types_map: Dict[str, Optional[str]] = {
'200': "str"
}
response_data = self.api_client.call_api(
*_param,
_request_timeout=_request_timeout
)
response_data.read()
return self.api_client.response_deserialize(
response_data=response_data,
response_types_map=_response_types_map,
)
@validate_call
def test_header_integer_boolean_string_enums_without_preload_content(
self,
integer_header: Optional[StrictInt] = None,
boolean_header: Optional[StrictBool] = None,
string_header: Optional[StrictStr] = None,
enum_nonref_string_header: Optional[StrictStr] = None,
enum_ref_string_header: Optional[StringEnumRef] = None,
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[
Annotated[StrictFloat, Field(gt=0)],
Annotated[StrictFloat, Field(gt=0)]
]
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> RESTResponseType:
"""Test header parameter(s)
Test header parameter(s)
:param integer_header:
:type integer_header: int
:param boolean_header:
:type boolean_header: bool
:param string_header:
:type string_header: str
:param enum_nonref_string_header:
:type enum_nonref_string_header: str
:param enum_ref_string_header:
:type enum_ref_string_header: StringEnumRef
: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.
:type _request_timeout: int, tuple(int, int), optional
: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.
:type _request_auth: dict, optional
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
""" # noqa: E501
_param = self._test_header_integer_boolean_string_enums_serialize(
integer_header=integer_header,
boolean_header=boolean_header,
string_header=string_header,
enum_nonref_string_header=enum_nonref_string_header,
enum_ref_string_header=enum_ref_string_header,
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index
)
_response_types_map: Dict[str, Optional[str]] = {
'200': "str"
}
response_data = self.api_client.call_api(
*_param,
_request_timeout=_request_timeout
)
return response_data.response
def _test_header_integer_boolean_string_enums_serialize(
self,
integer_header,
boolean_header,
string_header,
enum_nonref_string_header,
enum_ref_string_header,
_request_auth,
_content_type,
_headers,
_host_index,
) -> Tuple:
_host = None
_collection_formats: Dict[str, str] = {
}
_path_params: Dict[str, str] = {}
_query_params: List[Tuple[str, str]] = []
_header_params: Dict[str, Optional[str]] = _headers or {}
_form_params: List[Tuple[str, str]] = []
_files: Dict[str, str] = {}
_body_params: Optional[bytes] = None
# process the path parameters
# process the query parameters
# process the header parameters
if integer_header is not None:
_header_params['integer_header'] = integer_header
if boolean_header is not None:
_header_params['boolean_header'] = boolean_header
if string_header is not None:
_header_params['string_header'] = string_header
if enum_nonref_string_header is not None:
_header_params['enum_nonref_string_header'] = enum_nonref_string_header
if enum_ref_string_header is not None:
_header_params['enum_ref_string_header'] = enum_ref_string_header
# process the form parameters
# process the body parameter
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
'_request_auth',
'_content_type',
'_headers'
'text/plain'
]
)
# validate the arguments
for _key, _val in _params['kwargs'].items():
if _key not in _all_params:
raise ApiTypeError(
"Got an unexpected keyword argument '%s'"
" to method test_header_integer_boolean_string_enums" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats: Dict[str, str] = {}
# process the path parameters
_path_params: Dict[str, str] = {}
# process the query parameters
_query_params: List[Tuple[str, str]] = []
# process the header parameters
_header_params = dict(_params.get('_headers', {}))
if _params['integer_header'] is not None:
_header_params['integer_header'] = _params['integer_header']
if _params['boolean_header'] is not None:
_header_params['boolean_header'] = _params['boolean_header']
if _params['string_header'] is not None:
_header_params['string_header'] = _params['string_header']
if _params['enum_nonref_string_header'] is not None:
_header_params['enum_nonref_string_header'] = _params['enum_nonref_string_header']
if _params['enum_ref_string_header'] is not None:
_header_params['enum_ref_string_header'] = _params['enum_ref_string_header']
# process the form parameters
_form_params: List[Tuple[str, str]] = []
_files: Dict[str, str] = {}
# process the body parameter
_body_params = None
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
['text/plain']) # noqa: E501
# authentication setting
_auth_settings: List[str] = [] # noqa: E501
_auth_settings: List[str] = [
]
_response_types_map: Dict[str, Optional[str]] = {
'200': "str",
}
return self.api_client.call_api(
'/header/integer/boolean/string/enums', 'GET',
_path_params,
_query_params,
_header_params,
return self.api_client.param_serialize(
method='GET',
resource_path='/header/integer/boolean/string/enums',
path_params=_path_params,
query_params=_query_params,
header_params=_header_params,
body=_body_params,
post_params=_form_params,
files=_files,
response_types_map=_response_types_map,
auth_settings=_auth_settings,
async_req=_params.get('async_req'),
_return_http_data_only=_params.get('_return_http_data_only'), # noqa: E501
_preload_content=_params.get('_preload_content', True),
_request_timeout=_params.get('_request_timeout'),
collection_formats=_collection_formats,
_request_auth=_params.get('_request_auth'))
_host=_host,
_request_auth=_request_auth
)

384
samples/client/echo_api/python/openapi_client/api/path_api.py
View File

@@ -13,12 +13,16 @@
""" # noqa: E501
import re # noqa: F401
import io
import warnings
from pydantic import validate_call, ValidationError
from typing import Dict, List, Optional, Tuple
from pydantic import validate_call, Field, StrictFloat, StrictStr, StrictInt
from typing import Dict, List, Optional, Tuple, Union, Any
try:
from typing import Annotated
except ImportError:
from typing_extensions import Annotated
from pydantic import StrictInt, StrictStr, field_validator
@@ -26,10 +30,7 @@ from openapi_client.models.string_enum_ref import StringEnumRef
from openapi_client.api_client import ApiClient
from openapi_client.api_response import ApiResponse
from openapi_client.exceptions import ( # noqa: F401
ApiTypeError,
ApiValueError
)
from openapi_client.rest import RESTResponseType
class PathApi:
@@ -44,6 +45,7 @@ class PathApi:
api_client = ApiClient.get_default()
self.api_client = api_client
@validate_call
def tests_path_string_path_string_integer_path_integer_enum_nonref_string_path_enum_ref_string_path(
self,
@@ -51,16 +53,22 @@ class PathApi:
path_integer: StrictInt,
enum_nonref_string_path: StrictStr,
enum_ref_string_path: StringEnumRef,
**kwargs,
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[
Annotated[StrictFloat, Field(gt=0)],
Annotated[StrictFloat, Field(gt=0)]
]
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> str:
"""Test path parameter(s) # noqa: E501
"""Test path parameter(s)
Test path parameter(s) # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.tests_path_string_path_string_integer_path_integer_enum_nonref_string_path_enum_ref_string_path(path_string, path_integer, enum_nonref_string_path, enum_ref_string_path, async_req=True)
>>> result = thread.get()
Test path parameter(s)
:param path_string: (required)
:type path_string: str
@@ -70,30 +78,54 @@ class PathApi:
:type enum_nonref_string_path: str
:param enum_ref_string_path: (required)
:type enum_ref_string_path: StringEnumRef
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
: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_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.
:type _request_timeout: int, tuple(int, int), optional
: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.
:type _request_auth: dict, optional
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: str
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the tests_path_string_path_string_integer_path_integer_enum_nonref_string_path_enum_ref_string_path_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
""" # noqa: E501
return self.tests_path_string_path_string_integer_path_integer_enum_nonref_string_path_enum_ref_string_path_with_http_info.raw_function(
path_string,
path_integer,
enum_nonref_string_path,
enum_ref_string_path,
**kwargs,
_param = self._tests_path_string_path_string_integer_path_integer_enum_nonref_string_path_enum_ref_string_path_serialize(
path_string=path_string,
path_integer=path_integer,
enum_nonref_string_path=enum_nonref_string_path,
enum_ref_string_path=enum_ref_string_path,
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index
)
_response_types_map: Dict[str, Optional[str]] = {
'200': "str"
}
response_data = self.api_client.call_api(
*_param,
_request_timeout=_request_timeout
)
response_data.read()
return self.api_client.response_deserialize(
response_data=response_data,
response_types_map=_response_types_map,
).data
@validate_call
def tests_path_string_path_string_integer_path_integer_enum_nonref_string_path_enum_ref_string_path_with_http_info(
self,
@@ -101,16 +133,22 @@ class PathApi:
path_integer: StrictInt,
enum_nonref_string_path: StrictStr,
enum_ref_string_path: StringEnumRef,
**kwargs,
) -> ApiResponse:
"""Test path parameter(s) # noqa: E501
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[
Annotated[StrictFloat, Field(gt=0)],
Annotated[StrictFloat, Field(gt=0)]
]
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> ApiResponse[str]:
"""Test path parameter(s)
Test path parameter(s) # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.tests_path_string_path_string_integer_path_integer_enum_nonref_string_path_enum_ref_string_path_with_http_info(path_string, path_integer, enum_nonref_string_path, enum_ref_string_path, async_req=True)
>>> result = thread.get()
Test path parameter(s)
:param path_string: (required)
:type path_string: str
@@ -120,111 +158,195 @@ class PathApi:
:type enum_nonref_string_path: str
:param enum_ref_string_path: (required)
:type enum_ref_string_path: StringEnumRef
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _preload_content: if False, the ApiResponse.data will
be set to none and raw_data will store the
HTTP response body without reading/decoding.
Default is True.
:type _preload_content: bool, optional
:param _return_http_data_only: response data instead of ApiResponse
object with status code, headers, etc
:type _return_http_data_only: bool, optional
: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.
:type _request_timeout: int, tuple(int, int), optional
: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.
request; this effectively ignores the
authentication in the spec for a single request.
:type _request_auth: dict, optional
:type _content_type: string, optional: force content-type for the request
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: tuple(str, status_code(int), headers(HTTPHeaderDict))
"""
""" # noqa: E501
_params = locals()
_param = self._tests_path_string_path_string_integer_path_integer_enum_nonref_string_path_enum_ref_string_path_serialize(
path_string=path_string,
path_integer=path_integer,
enum_nonref_string_path=enum_nonref_string_path,
enum_ref_string_path=enum_ref_string_path,
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index
)
_all_params = [
'path_string',
'path_integer',
'enum_nonref_string_path',
'enum_ref_string_path'
]
_all_params.extend(
_response_types_map: Dict[str, Optional[str]] = {
'200': "str"
}
response_data = self.api_client.call_api(
*_param,
_request_timeout=_request_timeout
)
response_data.read()
return self.api_client.response_deserialize(
response_data=response_data,
response_types_map=_response_types_map,
)
@validate_call
def tests_path_string_path_string_integer_path_integer_enum_nonref_string_path_enum_ref_string_path_without_preload_content(
self,
path_string: StrictStr,
path_integer: StrictInt,
enum_nonref_string_path: StrictStr,
enum_ref_string_path: StringEnumRef,
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[
Annotated[StrictFloat, Field(gt=0)],
Annotated[StrictFloat, Field(gt=0)]
]
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> RESTResponseType:
"""Test path parameter(s)
Test path parameter(s)
:param path_string: (required)
:type path_string: str
:param path_integer: (required)
:type path_integer: int
:param enum_nonref_string_path: (required)
:type enum_nonref_string_path: str
:param enum_ref_string_path: (required)
:type enum_ref_string_path: StringEnumRef
: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.
:type _request_timeout: int, tuple(int, int), optional
: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.
:type _request_auth: dict, optional
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
""" # noqa: E501
_param = self._tests_path_string_path_string_integer_path_integer_enum_nonref_string_path_enum_ref_string_path_serialize(
path_string=path_string,
path_integer=path_integer,
enum_nonref_string_path=enum_nonref_string_path,
enum_ref_string_path=enum_ref_string_path,
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index
)
_response_types_map: Dict[str, Optional[str]] = {
'200': "str"
}
response_data = self.api_client.call_api(
*_param,
_request_timeout=_request_timeout
)
return response_data.response
def _tests_path_string_path_string_integer_path_integer_enum_nonref_string_path_enum_ref_string_path_serialize(
self,
path_string,
path_integer,
enum_nonref_string_path,
enum_ref_string_path,
_request_auth,
_content_type,
_headers,
_host_index,
) -> Tuple:
_host = None
_collection_formats: Dict[str, str] = {
}
_path_params: Dict[str, str] = {}
_query_params: List[Tuple[str, str]] = []
_header_params: Dict[str, Optional[str]] = _headers or {}
_form_params: List[Tuple[str, str]] = []
_files: Dict[str, str] = {}
_body_params: Optional[bytes] = None
# process the path parameters
if path_string is not None:
_path_params['path_string'] = path_string
if path_integer is not None:
_path_params['path_integer'] = path_integer
if enum_nonref_string_path is not None:
_path_params['enum_nonref_string_path'] = enum_nonref_string_path
if enum_ref_string_path is not None:
_path_params['enum_ref_string_path'] = enum_ref_string_path.value
# process the query parameters
# process the header parameters
# process the form parameters
# process the body parameter
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
'_request_auth',
'_content_type',
'_headers'
'text/plain'
]
)
# validate the arguments
for _key, _val in _params['kwargs'].items():
if _key not in _all_params:
raise ApiTypeError(
"Got an unexpected keyword argument '%s'"
" to method tests_path_string_path_string_integer_path_integer_enum_nonref_string_path_enum_ref_string_path" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats: Dict[str, str] = {}
# process the path parameters
_path_params: Dict[str, str] = {}
if _params['path_string'] is not None:
_path_params['path_string'] = _params['path_string']
if _params['path_integer'] is not None:
_path_params['path_integer'] = _params['path_integer']
if _params['enum_nonref_string_path'] is not None:
_path_params['enum_nonref_string_path'] = _params['enum_nonref_string_path']
if _params['enum_ref_string_path'] is not None:
_path_params['enum_ref_string_path'] = _params['enum_ref_string_path'].value
# process the query parameters
_query_params: List[Tuple[str, str]] = []
# process the header parameters
_header_params = dict(_params.get('_headers', {}))
# process the form parameters
_form_params: List[Tuple[str, str]] = []
_files: Dict[str, str] = {}
# process the body parameter
_body_params = None
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
['text/plain']) # noqa: E501
# authentication setting
_auth_settings: List[str] = [] # noqa: E501
_auth_settings: List[str] = [
]
_response_types_map: Dict[str, Optional[str]] = {
'200': "str",
}
return self.api_client.call_api(
'/path/string/{path_string}/integer/{path_integer}/{enum_nonref_string_path}/{enum_ref_string_path}', 'GET',
_path_params,
_query_params,
_header_params,
return self.api_client.param_serialize(
method='GET',
resource_path='/path/string/{path_string}/integer/{path_integer}/{enum_nonref_string_path}/{enum_ref_string_path}',
path_params=_path_params,
query_params=_query_params,
header_params=_header_params,
body=_body_params,
post_params=_form_params,
files=_files,
response_types_map=_response_types_map,
auth_settings=_auth_settings,
async_req=_params.get('async_req'),
_return_http_data_only=_params.get('_return_http_data_only'), # noqa: E501
_preload_content=_params.get('_preload_content', True),
_request_timeout=_params.get('_request_timeout'),
collection_formats=_collection_formats,
_request_auth=_params.get('_request_auth'))
_host=_host,
_request_auth=_request_auth
)

2751
samples/client/echo_api/python/openapi_client/api/query_api.py
View File

File diff suppressed because it is too large Load Diff

486
samples/client/echo_api/python/openapi_client/api_client.py
View File

@@ -18,18 +18,26 @@ import datetime
from dateutil.parser import parse
import json
import mimetypes
from multiprocessing.pool import ThreadPool
import os
import re
import tempfile
from urllib.parse import quote
from typing import Tuple, Optional, List
from openapi_client.configuration import Configuration
from openapi_client.api_response import ApiResponse
import openapi_client.models
from openapi_client import rest
from openapi_client.exceptions import ApiValueError, ApiException
from openapi_client.exceptions import (
ApiValueError,
ApiException,
BadRequestException,
UnauthorizedException,
ForbiddenException,
NotFoundException,
ServiceException
)
class ApiClient:
@@ -46,8 +54,6 @@ class ApiClient:
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)
@@ -63,13 +69,17 @@ class ApiClient:
}
_pool = None
def __init__(self, configuration=None, header_name=None, header_value=None,
cookie=None, pool_threads=1) -> None:
def __init__(
self,
configuration=None,
header_name=None,
header_value=None,
cookie=None
) -> None:
# use default configuration if none is provided
if configuration is None:
configuration = Configuration.get_default()
self.configuration = configuration
self.pool_threads = pool_threads
self.rest_client = rest.RESTClientObject(configuration)
self.default_headers = {}
@@ -80,29 +90,6 @@ class ApiClient:
self.user_agent = 'OpenAPI-Generator/1.0.0/python'
self.client_side_validation = configuration.client_side_validation
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 = ThreadPool(self.pool_threads)
return self._pool
@property
def user_agent(self):
@@ -143,13 +130,42 @@ class ApiClient:
"""
cls._default = default
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,
_return_http_data_only=None, collection_formats=None,
_preload_content=True, _request_timeout=None, _host=None,
_request_auth=None):
def param_serialize(
self,
method,
resource_path,
path_params=None,
query_params=None,
header_params=None,
body=None,
post_params=None,
files=None, auth_settings=None,
collection_formats=None,
_host=None,
_request_auth=None
) -> Tuple:
"""Builds the HTTP request params needed by the request.
:param method: Method to call.
:param resource_path: Path to method endpoint.
: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 files dict: key -> filename, value -> filepath,
for `multipart/form-data`.
:param collection_formats: dict of collection formats for path, query,
header, and post parameters.
: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.
:return: tuple of form (path, http_method, query_params, header_params,
body, post_params, files)
"""
config = self.configuration
@@ -160,14 +176,17 @@ class ApiClient:
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))
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)
path_params = self.parameters_to_tuples(
path_params,
collection_formats
)
for k, v in path_params:
# specified safe chars, encode everything
resource_path = resource_path.replace(
@@ -179,15 +198,22 @@ class ApiClient:
if post_params or files:
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)
post_params = self.parameters_to_tuples(
post_params,
collection_formats
)
post_params.extend(self.files_parameters(files))
# auth setting
self.update_params_for_auth(
header_params, query_params, auth_settings,
resource_path, method, body,
request_auth=_request_auth)
header_params,
query_params,
auth_settings,
resource_path,
method,
body,
request_auth=_request_auth
)
# body
if body:
@@ -203,59 +229,110 @@ class ApiClient:
# query parameters
if query_params:
query_params = self.sanitize_for_serialization(query_params)
url_query = self.parameters_to_url_query(query_params,
collection_formats)
url_query = self.parameters_to_url_query(
query_params,
collection_formats
)
url += "?" + url_query
return method, url, header_params, body, post_params
def call_api(
self,
method,
url,
header_params=None,
body=None,
post_params=None,
_request_timeout=None
) -> rest.RESTResponse:
"""Makes the HTTP request (synchronous)
:param method: Method to call.
:param url: Path to method endpoint.
: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 _request_timeout: timeout setting for this request.
:return: RESTResponse
"""
try:
# perform request and return response
response_data = self.request(
response_data = self.rest_client.request(
method, url,
query_params=query_params,
headers=header_params,
post_params=post_params, body=body,
_preload_content=_preload_content,
_request_timeout=_request_timeout)
body=body, post_params=post_params,
_request_timeout=_request_timeout
)
except ApiException as e:
if e.body:
e.body = e.body.decode('utf-8')
raise e
self.last_response = response_data
return response_data
return_data = None # assuming deserialization is not needed
# data needs deserialization or returns HTTP data (deserialized) only
if _preload_content or _return_http_data_only:
response_type = response_types_map.get(str(response_data.status), None)
if not response_type and isinstance(response_data.status, int) and 100 <= response_data.status <= 599:
# if not found, look for '1XX', '2XX', etc.
response_type = response_types_map.get(str(response_data.status)[0] + "XX", None)
def response_deserialize(
self,
response_data=None,
response_types_map=None
) -> ApiResponse:
"""Deserializes response into an object.
:param response_data: RESTResponse object to be deserialized.
:param response_types_map: dict of response types.
:return: ApiResponse
"""
if response_type == "bytearray":
response_data.data = response_data.data
else:
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"
response_data.data = response_data.data.decode(encoding)
# deserialize response data
if response_type == "bytearray":
return_data = response_data.data
elif response_type:
return_data = self.deserialize(response_data, response_type)
else:
return_data = None
response_type = response_types_map.get(str(response_data.status), None)
if not response_type and isinstance(response_data.status, int) and 100 <= response_data.status <= 599:
# if not found, look for '1XX', '2XX', etc.
response_type = response_types_map.get(str(response_data.status)[0] + "XX", None)
if _return_http_data_only:
return return_data
if response_type is None:
if not 200 <= response_data.status <= 299:
if response_data.status == 400:
raise BadRequestException(http_resp=response_data)
if response_data.status == 401:
raise UnauthorizedException(http_resp=response_data)
if response_data.status == 403:
raise ForbiddenException(http_resp=response_data)
if response_data.status == 404:
raise NotFoundException(http_resp=response_data)
if 500 <= response_data.status <= 599:
raise ServiceException(http_resp=response_data)
raise ApiException(http_resp=response_data)
# deserialize response data
if response_type == "bytearray":
return_data = response_data.data
elif response_type is None:
return_data = None
elif response_type == "file":
return_data = self.__deserialize_file(response_data)
else:
return ApiResponse(status_code = response_data.status,
data = return_data,
headers = response_data.getheaders(),
raw_data = response_data.data)
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"
response_text = response_data.data.decode(encoding)
return_data = self.deserialize(response_text, response_type)
return ApiResponse(
status_code = response_data.status,
data = return_data,
headers = response_data.getheaders(),
raw_data = response_data.data
)
def sanitize_for_serialization(self, obj):
"""Builds a JSON POST object.
@@ -276,15 +353,17 @@ class ApiClient:
elif isinstance(obj, self.PRIMITIVE_TYPES):
return obj
elif isinstance(obj, list):
return [self.sanitize_for_serialization(sub_obj)
for sub_obj in obj]
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)
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):
elif isinstance(obj, dict):
obj_dict = obj
else:
# Convert model obj to dict except
@@ -294,10 +373,12 @@ class ApiClient:
# model definition for request.
obj_dict = obj.to_dict()
return {key: self.sanitize_for_serialization(val)
for key, val in obj_dict.items()}
return {
key: self.sanitize_for_serialization(val)
for key, val in obj_dict.items()
}
def deserialize(self, response, response_type):
def deserialize(self, response_text, response_type):
"""Deserializes response into an object.
:param response: RESTResponse object to be deserialized.
@@ -306,16 +387,12 @@ class ApiClient:
: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)
data = json.loads(response_text)
except ValueError:
data = response.data
data = response_text
return self.__deserialize(data, response_type)
@@ -358,136 +435,6 @@ class ApiClient:
else:
return self.__deserialize_model(data, klass)
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):
"""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: key -> filename, value -> filepath,
for `multipart/form-data`.
:param async_req bool: execute request asynchronously
:param _return_http_data_only: response data instead of ApiResponse
object with status code, headers, etc
:param _preload_content: if False, the ApiResponse.data will
be set to none and raw_data will store the
HTTP response body without reading/decoding.
Default is True.
:param collection_formats: dict of collection formats for path, query,
header, and post parameters.
: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.
: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.
"""
args = (
resource_path,
method,
path_params,
query_params,
header_params,
body,
post_params,
files,
response_types_map,
auth_settings,
_return_http_data_only,
collection_formats,
_preload_content,
_request_timeout,
_host,
_request_auth,
)
if not async_req:
return self.__call_api(*args)
return self.pool.apply_async(self.__call_api, args)
def request(self, method, url, query_params=None, headers=None,
post_params=None, body=None, _preload_content=True,
_request_timeout=None):
"""Makes the HTTP request using RESTClient."""
if method == "GET":
return self.rest_client.get_request(url,
query_params=query_params,
_preload_content=_preload_content,
_request_timeout=_request_timeout,
headers=headers)
elif method == "HEAD":
return self.rest_client.head_request(url,
query_params=query_params,
_preload_content=_preload_content,
_request_timeout=_request_timeout,
headers=headers)
elif method == "OPTIONS":
return self.rest_client.options_request(url,
query_params=query_params,
headers=headers,
_preload_content=_preload_content,
_request_timeout=_request_timeout)
elif method == "POST":
return self.rest_client.post_request(url,
query_params=query_params,
headers=headers,
post_params=post_params,
_preload_content=_preload_content,
_request_timeout=_request_timeout,
body=body)
elif method == "PUT":
return self.rest_client.put_request(url,
query_params=query_params,
headers=headers,
post_params=post_params,
_preload_content=_preload_content,
_request_timeout=_request_timeout,
body=body)
elif method == "PATCH":
return self.rest_client.patch_request(url,
query_params=query_params,
headers=headers,
post_params=post_params,
_preload_content=_preload_content,
_request_timeout=_request_timeout,
body=body)
elif method == "DELETE":
return self.rest_client.delete_request(url,
query_params=query_params,
headers=headers,
_preload_content=_preload_content,
_request_timeout=_request_timeout,
body=body)
else:
raise ApiValueError(
"http method must be `GET`, `HEAD`, `OPTIONS`,"
" `POST`, `PATCH`, `PUT` or `DELETE`."
)
def parameters_to_tuples(self, params, collection_formats):
"""Get parameters as list of tuples, formatting collections.
@@ -498,7 +445,7 @@ class ApiClient:
new_params = []
if collection_formats is None:
collection_formats = {}
for k, v in params.items() if isinstance(params, dict) else params: # noqa: E501
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':
@@ -528,7 +475,7 @@ class ApiClient:
new_params = []
if collection_formats is None:
collection_formats = {}
for k, v in params.items() if isinstance(params, dict) else params: # noqa: E501
for k, v in params.items() if isinstance(params, dict) else params:
if isinstance(v, (int, float)):
v = str(v)
if isinstance(v, bool):
@@ -550,7 +497,8 @@ class ApiClient:
else: # csv is the default
delimiter = ','
new_params.append(
(k, delimiter.join(quote(str(value)) for value in v)))
(k, delimiter.join(quote(str(value)) for value in v))
)
else:
new_params.append((k, quote(str(v))))
@@ -573,21 +521,24 @@ class ApiClient:
with open(n, 'rb') as f:
filename = os.path.basename(f.name)
filedata = f.read()
mimetype = (mimetypes.guess_type(filename)[0] or
'application/octet-stream')
mimetype = (
mimetypes.guess_type(filename)[0]
or 'application/octet-stream'
)
params.append(
tuple([k, tuple([filename, filedata, mimetype])]))
tuple([k, tuple([filename, filedata, mimetype])])
)
return params
def select_header_accept(self, accepts):
def select_header_accept(self, accepts: List[str]) -> Optional[str]:
"""Returns `Accept` based on an array of accepts provided.
:param accepts: List of headers.
:return: Accept (e.g. application/json).
"""
if not accepts:
return
return None
for accept in accepts:
if re.search('json', accept, re.IGNORECASE):
@@ -610,9 +561,16 @@ class ApiClient:
return content_types[0]
def update_params_for_auth(self, headers, queries, auth_settings,
resource_path, method, body,
request_auth=None):
def update_params_for_auth(
self,
headers,
queries,
auth_settings,
resource_path,
method,
body,
request_auth=None
) -> None:
"""Updates header and query params based on authentication setting.
:param headers: Header parameters dict to be updated.
@@ -629,21 +587,36 @@ class ApiClient:
return
if request_auth:
self._apply_auth_params(headers, queries,
resource_path, method, body,
request_auth)
return
self._apply_auth_params(
headers,
queries,
resource_path,
method,
body,
request_auth
)
else:
for auth in auth_settings:
auth_setting = self.configuration.auth_settings().get(auth)
if auth_setting:
self._apply_auth_params(
headers,
queries,
resource_path,
method,
body,
auth_setting
)
for auth in auth_settings:
auth_setting = self.configuration.auth_settings().get(auth)
if auth_setting:
self._apply_auth_params(headers, queries,
resource_path, method, body,
auth_setting)
def _apply_auth_params(self, headers, queries,
resource_path, method, body,
auth_setting):
def _apply_auth_params(
self,
headers,
queries,
resource_path,
method,
body,
auth_setting
) -> None:
"""Updates the request parameters based on a single auth_setting
:param headers: Header parameters dict to be updated.
@@ -672,6 +645,9 @@ class ApiClient:
Saves response body into a file in a temporary folder,
using the filename from the `Content-Disposition` header if provided.
handle file downloading
save response body into a tmp file and return the instance
:param response: RESTResponse.
:return: file path.
"""
@@ -681,8 +657,10 @@ class ApiClient:
content_disposition = response.getheader("Content-Disposition")
if content_disposition:
filename = re.search(r'filename=[\'"]?([^\'"\s]+)[\'"]?',
content_disposition).group(1)
filename = re.search(
r'filename=[\'"]?([^\'"\s]+)[\'"]?',
content_disposition
).group(1)
path = os.path.join(os.path.dirname(path), filename)
with open(path, "wb") as f:

26
samples/client/echo_api/python/openapi_client/api_response.py
View File

@@ -1,25 +1,21 @@
"""API response object."""
from __future__ import annotations
from typing import Any, Dict, Optional
from pydantic import Field, StrictInt, StrictStr
from typing import Any, Dict, Optional, Generic, TypeVar
from pydantic import Field, StrictInt, StrictStr, StrictBytes, BaseModel
class ApiResponse:
T = TypeVar("T")
class ApiResponse(BaseModel, Generic[T]):
"""
API response object
"""
status_code: Optional[StrictInt] = Field(None, description="HTTP status code")
status_code: StrictInt = Field(description="HTTP status code")
headers: Optional[Dict[StrictStr, StrictStr]] = Field(None, description="HTTP headers")
data: Optional[Any] = Field(None, description="Deserialized data given the data type")
raw_data: Optional[Any] = Field(None, description="Raw data (HTTP response body)")
data: T = Field(description="Deserialized data given the data type")
raw_data: StrictBytes = Field(description="Raw data (HTTP response body)")
def __init__(self,
status_code=None,
headers=None,
data=None,
raw_data=None) -> None:
self.status_code = status_code
self.headers = headers
self.data = data
self.raw_data = raw_data
model_config = {
"arbitrary_types_allowed": True
}

8
samples/client/echo_api/python/openapi_client/configuration.py
View File

@@ -15,7 +15,6 @@
import copy
import logging
import multiprocessing
import sys
import urllib3
@@ -165,13 +164,6 @@ conf = openapi_client.Configuration(
Set this to the SNI value expected by the server.
"""
self.connection_pool_maxsize = multiprocessing.cpu_count() * 5
"""urllib3 connection pool's maximum number of connections saved
per pool. urllib3 uses 1 connection as default value, but this is
not the best value when you are making a lot of possibly parallel
requests to the same host, which is often the case here.
cpu_count * 5 is used as default value to increase performance.
"""
self.proxy = None
"""Proxy URL

2
samples/client/echo_api/python/openapi_client/exceptions.py
View File

@@ -106,7 +106,7 @@ class ApiException(OpenApiException):
if http_resp:
self.status = http_resp.status
self.reason = http_resp.reason
self.body = http_resp.data
self.body = http_resp.data.decode('utf-8')
self.headers = http_resp.getheaders()
else:
self.status = status

7
samples/client/echo_api/python/openapi_client/models/bird.py
View File

@@ -19,9 +19,8 @@ import re # noqa: F401
import json
from typing import Optional
from typing import Any, ClassVar, Dict, List, Optional
from pydantic import BaseModel, StrictStr
from typing import Dict, Any
try:
from typing import Self
except ImportError:
@@ -30,7 +29,7 @@ except ImportError:
class Bird(BaseModel):
"""
Bird
"""
""" # noqa: E501
size: Optional[StrictStr] = None
color: Optional[StrictStr] = None
__properties: ClassVar[List[str]] = ["size", "color"]
@@ -74,7 +73,7 @@ class Bird(BaseModel):
return _dict
@classmethod
def from_dict(cls, obj: dict) -> Self:
def from_dict(cls, obj: Dict) -> Self:
"""Create an instance of Bird from a dict"""
if obj is None:
return None

7
samples/client/echo_api/python/openapi_client/models/category.py
View File

@@ -19,9 +19,8 @@ import re # noqa: F401
import json
from typing import Optional
from typing import Any, ClassVar, Dict, List, Optional
from pydantic import BaseModel, StrictInt, StrictStr
from typing import Dict, Any
try:
from typing import Self
except ImportError:
@@ -30,7 +29,7 @@ except ImportError:
class Category(BaseModel):
"""
Category
"""
""" # noqa: E501
id: Optional[StrictInt] = None
name: Optional[StrictStr] = None
__properties: ClassVar[List[str]] = ["id", "name"]
@@ -74,7 +73,7 @@ class Category(BaseModel):
return _dict
@classmethod
def from_dict(cls, obj: dict) -> Self:
def from_dict(cls, obj: Dict) -> Self:
"""Create an instance of Category from a dict"""
if obj is None:
return None

7
samples/client/echo_api/python/openapi_client/models/data_query.py
View File

@@ -19,11 +19,10 @@ import re # noqa: F401
import json
from datetime import datetime
from typing import Optional
from typing import Any, ClassVar, Dict, List, Optional
from pydantic import StrictStr
from pydantic import Field
from openapi_client.models.query import Query
from typing import Dict, Any
try:
from typing import Self
except ImportError:
@@ -32,7 +31,7 @@ except ImportError:
class DataQuery(Query):
"""
DataQuery
"""
""" # noqa: E501
suffix: Optional[StrictStr] = Field(default=None, description="test suffix")
text: Optional[StrictStr] = Field(default=None, description="Some text containing white spaces")
var_date: Optional[datetime] = Field(default=None, description="A date", alias="date")
@@ -77,7 +76,7 @@ class DataQuery(Query):
return _dict
@classmethod
def from_dict(cls, obj: dict) -> Self:
def from_dict(cls, obj: Dict) -> Self:
"""Create an instance of DataQuery from a dict"""
if obj is None:
return None

9
samples/client/echo_api/python/openapi_client/models/default_value.py
View File

@@ -19,10 +19,9 @@ import re # noqa: F401
import json
from typing import List, Optional
from typing import Any, ClassVar, Dict, List, Optional
from pydantic import BaseModel, StrictInt, StrictStr, field_validator
from openapi_client.models.string_enum_ref import StringEnumRef
from typing import Dict, Any
try:
from typing import Self
except ImportError:
@@ -30,8 +29,8 @@ except ImportError:
class DefaultValue(BaseModel):
"""
to test the default value of properties # noqa: E501
"""
to test the default value of properties
""" # noqa: E501
array_string_enum_ref_default: Optional[List[StringEnumRef]] = None
array_string_enum_default: Optional[List[StrictStr]] = None
array_string_default: Optional[List[StrictStr]] = None
@@ -107,7 +106,7 @@ class DefaultValue(BaseModel):
return _dict
@classmethod
def from_dict(cls, obj: dict) -> Self:
def from_dict(cls, obj: Dict) -> Self:
"""Create an instance of DefaultValue from a dict"""
if obj is None:
return None

7
samples/client/echo_api/python/openapi_client/models/number_properties_only.py
View File

@@ -19,11 +19,10 @@ import re # noqa: F401
import json
from typing import Optional, Union
from typing import Any, ClassVar, Dict, List, Optional, Union
from pydantic import BaseModel, StrictFloat, StrictInt
from pydantic import Field
from typing_extensions import Annotated
from typing import Dict, Any
try:
from typing import Self
except ImportError:
@@ -32,7 +31,7 @@ except ImportError:
class NumberPropertiesOnly(BaseModel):
"""
NumberPropertiesOnly
"""
""" # noqa: E501
number: Optional[Union[StrictFloat, StrictInt]] = None
var_float: Optional[Union[StrictFloat, StrictInt]] = Field(default=None, alias="float")
double: Optional[Union[Annotated[float, Field(le=50.2, strict=True, ge=0.8)], Annotated[int, Field(le=50, strict=True, ge=1)]]] = None
@@ -77,7 +76,7 @@ class NumberPropertiesOnly(BaseModel):
return _dict
@classmethod
def from_dict(cls, obj: dict) -> Self:
def from_dict(cls, obj: Dict) -> Self:
"""Create an instance of NumberPropertiesOnly from a dict"""
if obj is None:
return None

7
samples/client/echo_api/python/openapi_client/models/pet.py
View File

@@ -19,12 +19,11 @@ import re # noqa: F401
import json
from typing import List, Optional
from typing import Any, ClassVar, Dict, List, Optional
from pydantic import BaseModel, StrictInt, StrictStr, field_validator
from pydantic import Field
from openapi_client.models.category import Category
from openapi_client.models.tag import Tag
from typing import Dict, Any
try:
from typing import Self
except ImportError:
@@ -33,7 +32,7 @@ except ImportError:
class Pet(BaseModel):
"""
Pet
"""
""" # noqa: E501
id: Optional[StrictInt] = None
name: StrictStr
category: Optional[Category] = None
@@ -101,7 +100,7 @@ class Pet(BaseModel):
return _dict
@classmethod
def from_dict(cls, obj: dict) -> Self:
def from_dict(cls, obj: Dict) -> Self:
"""Create an instance of Pet from a dict"""
if obj is None:
return None

7
samples/client/echo_api/python/openapi_client/models/query.py
View File

@@ -19,10 +19,9 @@ import re # noqa: F401
import json
from typing import List, Optional
from typing import Any, ClassVar, Dict, List, Optional
from pydantic import BaseModel, StrictInt, StrictStr, field_validator
from pydantic import Field
from typing import Dict, Any
try:
from typing import Self
except ImportError:
@@ -31,7 +30,7 @@ except ImportError:
class Query(BaseModel):
"""
Query
"""
""" # noqa: E501
id: Optional[StrictInt] = Field(default=None, description="Query")
outcomes: Optional[List[StrictStr]] = None
__properties: ClassVar[List[str]] = ["id", "outcomes"]
@@ -86,7 +85,7 @@ class Query(BaseModel):
return _dict
@classmethod
def from_dict(cls, obj: dict) -> Self:
def from_dict(cls, obj: Dict) -> Self:
"""Create an instance of Query from a dict"""

7
samples/client/echo_api/python/openapi_client/models/tag.py
View File

@@ -19,9 +19,8 @@ import re # noqa: F401
import json
from typing import Optional
from typing import Any, ClassVar, Dict, List, Optional
from pydantic import BaseModel, StrictInt, StrictStr
from typing import Dict, Any
try:
from typing import Self
except ImportError:
@@ -30,7 +29,7 @@ except ImportError:
class Tag(BaseModel):
"""
Tag
"""
""" # noqa: E501
id: Optional[StrictInt] = None
name: Optional[StrictStr] = None
__properties: ClassVar[List[str]] = ["id", "name"]
@@ -74,7 +73,7 @@ class Tag(BaseModel):
return _dict
@classmethod
def from_dict(cls, obj: dict) -> Self:
def from_dict(cls, obj: Dict) -> Self:
"""Create an instance of Tag from a dict"""
if obj is None:
return None

7
samples/client/echo_api/python/openapi_client/models/test_query_style_deep_object_explode_true_object_all_of_query_object_parameter.py
View File

@@ -19,9 +19,8 @@ import re # noqa: F401
import json
from typing import Optional
from typing import Any, ClassVar, Dict, List, Optional
from pydantic import BaseModel, StrictInt, StrictStr
from typing import Dict, Any
try:
from typing import Self
except ImportError:
@@ -30,7 +29,7 @@ except ImportError:
class TestQueryStyleDeepObjectExplodeTrueObjectAllOfQueryObjectParameter(BaseModel):
"""
TestQueryStyleDeepObjectExplodeTrueObjectAllOfQueryObjectParameter
"""
""" # noqa: E501
size: Optional[StrictStr] = None
color: Optional[StrictStr] = None
id: Optional[StrictInt] = None
@@ -76,7 +75,7 @@ class TestQueryStyleDeepObjectExplodeTrueObjectAllOfQueryObjectParameter(BaseMod
return _dict
@classmethod
def from_dict(cls, obj: dict) -> Self:
def from_dict(cls, obj: Dict) -> Self:
"""Create an instance of TestQueryStyleDeepObjectExplodeTrueObjectAllOfQueryObjectParameter from a dict"""
if obj is None:
return None

7
samples/client/echo_api/python/openapi_client/models/test_query_style_form_explode_true_array_string_query_object_parameter.py
View File

@@ -19,9 +19,8 @@ import re # noqa: F401
import json
from typing import List, Optional
from typing import Any, ClassVar, Dict, List, Optional
from pydantic import BaseModel, StrictStr
from typing import Dict, Any
try:
from typing import Self
except ImportError:
@@ -30,7 +29,7 @@ except ImportError:
class TestQueryStyleFormExplodeTrueArrayStringQueryObjectParameter(BaseModel):
"""
TestQueryStyleFormExplodeTrueArrayStringQueryObjectParameter
"""
""" # noqa: E501
values: Optional[List[StrictStr]] = None
__properties: ClassVar[List[str]] = ["values"]
@@ -73,7 +72,7 @@ class TestQueryStyleFormExplodeTrueArrayStringQueryObjectParameter(BaseModel):
return _dict
@classmethod
def from_dict(cls, obj: dict) -> Self:
def from_dict(cls, obj: Dict) -> Self:
"""Create an instance of TestQueryStyleFormExplodeTrueArrayStringQueryObjectParameter from a dict"""
if obj is None:
return None

228
samples/client/echo_api/python/openapi_client/rest.py
View File

@@ -15,43 +15,43 @@
import io
import json
import logging
import re
import ssl
from urllib.parse import urlencode, quote_plus
import urllib3
from openapi_client.exceptions import ApiException, UnauthorizedException, ForbiddenException, NotFoundException, ServiceException, ApiValueError, BadRequestException
logger = logging.getLogger(__name__)
from openapi_client.exceptions import ApiException, ApiValueError
RESTResponseType = urllib3.HTTPResponse
class RESTResponse(io.IOBase):
def __init__(self, resp) -> None:
self.urllib3_response = resp
self.response = resp
self.status = resp.status
self.reason = resp.reason
self.data = resp.data
self.data = None
def read(self):
if self.data is None:
self.data = self.response.data
return self.data
def getheaders(self):
"""Returns a dictionary of the response headers."""
return self.urllib3_response.headers
return self.response.headers
def getheader(self, name, default=None):
"""Returns a given response header."""
return self.urllib3_response.headers.get(name, default)
return self.response.headers.get(name, default)
class RESTClientObject:
def __init__(self, configuration, pools_size=4, maxsize=None) -> None:
def __init__(self, configuration) -> None:
# urllib3.PoolManager will pass all kw parameters to connectionpool
# https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/poolmanager.py#L75 # noqa: E501
# https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/connectionpool.py#L680 # noqa: E501
# maxsize is the number of requests to host that are allowed in parallel # noqa: E501
# Custom SSL certificates and client certificates: http://urllib3.readthedocs.io/en/latest/advanced-usage.html # noqa: E501
# cert_reqs
@@ -62,7 +62,9 @@ class RESTClientObject:
addition_pool_args = {}
if configuration.assert_hostname is not None:
addition_pool_args['assert_hostname'] = configuration.assert_hostname # noqa: E501
addition_pool_args['assert_hostname'] = (
configuration.assert_hostname
)
if configuration.retries is not None:
addition_pool_args['retries'] = configuration.retries
@@ -74,17 +76,9 @@ class RESTClientObject:
if configuration.socket_options is not None:
addition_pool_args['socket_options'] = configuration.socket_options
if maxsize is None:
if configuration.connection_pool_maxsize is not None:
maxsize = configuration.connection_pool_maxsize
else:
maxsize = 4
# https pool manager
if configuration.proxy:
self.pool_manager = urllib3.ProxyManager(
num_pools=pools_size,
maxsize=maxsize,
cert_reqs=cert_reqs,
ca_certs=configuration.ssl_ca_cert,
cert_file=configuration.cert_file,
@@ -95,8 +89,6 @@ class RESTClientObject:
)
else:
self.pool_manager = urllib3.PoolManager(
num_pools=pools_size,
maxsize=maxsize,
cert_reqs=cert_reqs,
ca_certs=configuration.ssl_ca_cert,
cert_file=configuration.cert_file,
@@ -104,30 +96,39 @@ class RESTClientObject:
**addition_pool_args
)
def request(self, method, url, query_params=None, headers=None,
body=None, post_params=None, _preload_content=True,
_request_timeout=None):
def request(
self,
method,
url,
headers=None,
body=None,
post_params=None,
_request_timeout=None
):
"""Perform requests.
:param method: http request method
:param url: http request url
:param query_params: query parameters in the url
:param headers: http request headers
:param body: request json body, for `application/json`
:param post_params: request post parameters,
`application/x-www-form-urlencoded`
and `multipart/form-data`
: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.
"""
method = method.upper()
assert method in ['GET', 'HEAD', 'DELETE', 'POST', 'PUT',
'PATCH', 'OPTIONS']
assert method in [
'GET',
'HEAD',
'DELETE',
'POST',
'PUT',
'PATCH',
'OPTIONS'
]
if post_params and body:
raise ApiValueError(
@@ -136,65 +137,78 @@ class RESTClientObject:
post_params = post_params or {}
headers = headers or {}
# url already contains the URL query string
# so reset query_params to empty dict
query_params = {}
timeout = None
if _request_timeout:
if isinstance(_request_timeout, (int,float)): # noqa: E501,F821
if isinstance(_request_timeout, (int, float)):
timeout = urllib3.Timeout(total=_request_timeout)
elif (isinstance(_request_timeout, tuple) and
len(_request_timeout) == 2):
elif (
isinstance(_request_timeout, tuple)
and len(_request_timeout) == 2
):
timeout = urllib3.Timeout(
connect=_request_timeout[0], read=_request_timeout[1])
connect=_request_timeout[0],
read=_request_timeout[1]
)
try:
# For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE`
if method in ['POST', 'PUT', 'PATCH', 'OPTIONS', 'DELETE']:
# no content type provided or payload is json
if not headers.get('Content-Type') or re.search('json', headers['Content-Type'], re.IGNORECASE):
content_type = headers.get('Content-Type')
if (
not content_type
or re.search('json', content_type, re.IGNORECASE)
):
request_body = None
if body is not None:
request_body = json.dumps(body)
r = self.pool_manager.request(
method, url,
method,
url,
body=request_body,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
elif headers['Content-Type'] == 'application/x-www-form-urlencoded': # noqa: E501
headers=headers,
preload_content=False
)
elif content_type == 'application/x-www-form-urlencoded':
r = self.pool_manager.request(
method, url,
method,
url,
fields=post_params,
encode_multipart=False,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
elif headers['Content-Type'] == 'multipart/form-data':
headers=headers,
preload_content=False
)
elif content_type == 'multipart/form-data':
# must del headers['Content-Type'], or the correct
# Content-Type which generated by urllib3 will be
# overwritten.
del headers['Content-Type']
r = self.pool_manager.request(
method, url,
method,
url,
fields=post_params,
encode_multipart=True,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
headers=headers,
preload_content=False
)
# Pass a `string` parameter directly in the body to support
# other content types than Json when `body` argument is
# provided in serialized form
elif isinstance(body, str) or isinstance(body, bytes):
request_body = body
r = self.pool_manager.request(
method, url,
method,
url,
body=request_body,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
headers=headers,
preload_content=False
)
else:
# Cannot generate the request from given parameters
msg = """Cannot prepare a request message for provided
@@ -203,102 +217,16 @@ class RESTClientObject:
raise ApiException(status=0, reason=msg)
# For `GET`, `HEAD`
else:
r = self.pool_manager.request(method, url,
fields={},
preload_content=_preload_content,
timeout=timeout,
headers=headers)
r = self.pool_manager.request(
method,
url,
fields={},
timeout=timeout,
headers=headers,
preload_content=False
)
except urllib3.exceptions.SSLError as e:
msg = "{0}\n{1}".format(type(e).__name__, str(e))
msg = "\n".join([type(e).__name__, str(e)])
raise ApiException(status=0, reason=msg)
if _preload_content:
r = RESTResponse(r)
# log response body
logger.debug("response body: %s", r.data)
if not 200 <= r.status <= 299:
if r.status == 400:
raise BadRequestException(http_resp=r)
if r.status == 401:
raise UnauthorizedException(http_resp=r)
if r.status == 403:
raise ForbiddenException(http_resp=r)
if r.status == 404:
raise NotFoundException(http_resp=r)
if 500 <= r.status <= 599:
raise ServiceException(http_resp=r)
raise ApiException(http_resp=r)
return r
def get_request(self, url, headers=None, query_params=None, _preload_content=True,
_request_timeout=None):
return self.request("GET", url,
headers=headers,
_preload_content=_preload_content,
_request_timeout=_request_timeout,
query_params=query_params)
def head_request(self, url, headers=None, query_params=None, _preload_content=True,
_request_timeout=None):
return self.request("HEAD", url,
headers=headers,
_preload_content=_preload_content,
_request_timeout=_request_timeout,
query_params=query_params)
def options_request(self, url, headers=None, query_params=None, post_params=None,
body=None, _preload_content=True, _request_timeout=None):
return self.request("OPTIONS", url,
headers=headers,
query_params=query_params,
post_params=post_params,
_preload_content=_preload_content,
_request_timeout=_request_timeout,
body=body)
def delete_request(self, url, headers=None, query_params=None, body=None,
_preload_content=True, _request_timeout=None):
return self.request("DELETE", url,
headers=headers,
query_params=query_params,
_preload_content=_preload_content,
_request_timeout=_request_timeout,
body=body)
def post_request(self, url, headers=None, query_params=None, post_params=None,
body=None, _preload_content=True, _request_timeout=None):
return self.request("POST", url,
headers=headers,
query_params=query_params,
post_params=post_params,
_preload_content=_preload_content,
_request_timeout=_request_timeout,
body=body)
def put_request(self, url, headers=None, query_params=None, post_params=None,
body=None, _preload_content=True, _request_timeout=None):
return self.request("PUT", url,
headers=headers,
query_params=query_params,
post_params=post_params,
_preload_content=_preload_content,
_request_timeout=_request_timeout,
body=body)
def patch_request(self, url, headers=None, query_params=None, post_params=None,
body=None, _preload_content=True, _request_timeout=None):
return self.request("PATCH", url,
headers=headers,
query_params=query_params,
post_params=post_params,
_preload_content=_preload_content,
_request_timeout=_request_timeout,
body=body)
return RESTResponse(r)