loafle/openapi-generator-original
3
0
Fork 1
mirror of https://github.com/OpenAPITools/openapi-generator.git synced 2025-07-03 06:00:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

python: remove non-async code path from the aiohttp generator (#16601)

Browse Source 
* python: remove non-async code path from the aiohttp generator

This removes all the non-async code from the aiohttp generator:

* all the methods that should be asynchronous are marked as `async`
* the `async_req` parameter is gone; calls are directly awaitables now
* the async calls into a thread pool are gone and the thread pool is
  gone too (now useless)

Closes: #15824
Closes: #5539
Related: #763
Related: #3696

* Fix empty line

* remove more
This commit is contained in:
Jonathan Ballet 2023-09-20 04:48:31 +02:00 committed by GitHub
parent 69bc994ba7
commit a2f6b8eae5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 341 additions and 1513 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
modules/openapi-generator/src/main/resources/python/api.mustache vendored
View File

@ -37,35 +37,34 @@ class {{classname}}:
self.api_client = api_client
{{#operation}}
{{#asyncio}}
@overload
async def {{operationId}}(self, {{#allParams}}{{paramName}} : {{{vendorExtensions.x-py-typing}}}{{^required}} = None{{/required}}, {{/allParams}}**kwargs) -> {{{returnType}}}{{^returnType}}None{{/returnType}}: # noqa: E501
...
@overload
def {{operationId}}(self, {{#allParams}}{{paramName}} : {{{vendorExtensions.x-py-typing}}}{{^required}} = None{{/required}}, {{/allParams}}async_req: Optional[bool]=True, **kwargs) -> {{{returnType}}}{{^returnType}}None{{/returnType}}: # noqa: E501
...
{{/asyncio}}
@validate_arguments
def {{operationId}}(self, {{#allParams}}{{paramName}} : {{{vendorExtensions.x-py-typing}}}{{^required}} = None{{/required}}, {{/allParams}}{{#asyncio}}async_req: Optional[bool]=None, {{/asyncio}}**kwargs) -> {{#asyncio}}Union[{{{returnType}}}{{^returnType}}None{{/returnType}}, Awaitable[{{{returnType}}}{{^returnType}}None{{/returnType}}]]{{/asyncio}}{{^asyncio}}{{{returnType}}}{{^returnType}}None{{/returnType}}{{/asyncio}}: # noqa: E501
{{#asyncio}}
async def {{operationId}}(self, {{#allParams}}{{paramName}} : {{{vendorExtensions.x-py-typing}}}{{^required}} = None{{/required}}, {{/allParams}}**kwargs) -> {{{returnType}}}{{^returnType}}None{{/returnType}}: # noqa: E501
{{/asyncio}}
{{^asyncio}}
def {{operationId}}(self, {{#allParams}}{{paramName}} : {{{vendorExtensions.x-py-typing}}}{{^required}} = None{{/required}}, {{/allParams}}**kwargs) -> {{{returnType}}}{{^returnType}}None{{/returnType}}: # noqa: E501
{{/asyncio}}
"""{{#isDeprecated}}(Deprecated) {{/isDeprecated}}{{{summary}}}{{^summary}}{{operationId}}{{/summary}} # noqa: E501
{{#notes}}
{{{.}}} # noqa: E501
{{/notes}}
{{^asyncio}}
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.{{operationId}}({{#allParams}}{{paramName}}, {{/allParams}}async_req=True)
>>> result = thread.get()
{{/asyncio}}
{{#allParams}}
:param {{paramName}}:{{#description}} {{{.}}}{{/description}}{{#required}} (required){{/required}}{{#optional}}(optional){{/optional}}
:type {{paramName}}: {{dataType}}{{#optional}}, optional{{/optional}}
{{/allParams}}
{{^asyncio}}
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
{{/asyncio}}
: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
@ -79,31 +78,36 @@ class {{classname}}:
if '_preload_content' in kwargs:
message = "Error! Please call the {{operationId}}_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
{{#asyncio}}
if async_req is not None:
kwargs['async_req'] = async_req
{{/asyncio}}
return self.{{operationId}}_with_http_info({{#allParams}}{{paramName}}, {{/allParams}}**kwargs) # noqa: E501
return {{#asyncio}}await {{/asyncio}}self.{{operationId}}_with_http_info({{#allParams}}{{paramName}}, {{/allParams}}**kwargs) # noqa: E501
@validate_arguments
{{#asyncio}}
async def {{operationId}}_with_http_info(self, {{#allParams}}{{paramName}} : {{{vendorExtensions.x-py-typing}}}{{^required}} = None{{/required}}, {{/allParams}}**kwargs) -> ApiResponse: # noqa: E501
{{/asyncio}}
{{^asyncio}}
def {{operationId}}_with_http_info(self, {{#allParams}}{{paramName}} : {{{vendorExtensions.x-py-typing}}}{{^required}} = None{{/required}}, {{/allParams}}**kwargs) -> ApiResponse: # noqa: E501
{{/asyncio}}
"""{{#isDeprecated}}(Deprecated) {{/isDeprecated}}{{{summary}}}{{^summary}}{{operationId}}{{/summary}} # noqa: E501
{{#notes}}
{{{.}}} # noqa: E501
{{/notes}}
{{^asyncio}}
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.{{operationId}}_with_http_info({{#allParams}}{{paramName}}, {{/allParams}}async_req=True)
>>> result = thread.get()
{{/asyncio}}
{{#allParams}}
:param {{paramName}}:{{#description}} {{{.}}}{{/description}}{{#required}} (required){{/required}}{{#optional}}(optional){{/optional}}
:type {{paramName}}: {{dataType}}{{#optional}}, optional{{/optional}}
{{/allParams}}
{{^asyncio}}
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
{{/asyncio}}
: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.
@ -156,7 +160,9 @@ class {{classname}}:
]
_all_params.extend(
[
{{^asyncio}}
'async_req',
{{/asyncio}}
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -290,7 +296,7 @@ class {{classname}}:
_response_types_map = {}
{{/returnType}}
return self.api_client.call_api(
return {{#asyncio}}await {{/asyncio}}self.api_client.call_api(
'{{{path}}}', '{{httpMethod}}',
_path_params,
_query_params,
@ -300,7 +306,9 @@ class {{classname}}:
files=_files,
response_types_map=_response_types_map,
auth_settings=_auth_settings,
{{^asyncio}}
async_req=_params.get('async_req'),
{{/asyncio}}
_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'),

99
modules/openapi-generator/src/main/resources/python/api_client.mustache vendored
View File

@ -7,7 +7,9 @@ import datetime
from dateutil.parser import parse
import json
import mimetypes
{{^asyncio}}
from multiprocessing.pool import ThreadPool
{{/asyncio}}
import os
import re
import tempfile
@ -38,8 +40,10 @@ class ApiClient:
the API.
:param cookie: a cookie to include in the header when making calls
to the API
{{^asyncio}}
:param pool_threads: The number of threads to use for async requests
to the API. More threads means more concurrent API requests.
{{/asyncio}}
"""
PRIMITIVE_TYPES = (float, bool, bytes, str, int)
@ -56,12 +60,14 @@ class ApiClient:
_pool = None
def __init__(self, configuration=None, header_name=None, header_value=None,
cookie=None, pool_threads=1) -> None:
cookie=None{{^asyncio}}, pool_threads=1{{/asyncio}}) -> None:
# use default configuration if none is provided
if configuration is None:
configuration = Configuration.get_default()
self.configuration = configuration
{{^asyncio}}
self.pool_threads = pool_threads
{{/asyncio}}
self.rest_client = rest.RESTClientObject(configuration)
self.default_headers = {}
@ -72,25 +78,24 @@ class ApiClient:
self.user_agent = '{{{httpUserAgent}}}{{^httpUserAgent}}OpenAPI-Generator/{{{packageVersion}}}/python{{/httpUserAgent}}'
self.client_side_validation = configuration.client_side_validation
{{#asyncio}}
{{#asyncio}}
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_value, traceback):
await self.close()
{{/asyncio}}
{{^asyncio}}
async def close(self):
await self.rest_client.close()
{{/asyncio}}
{{^asyncio}}
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
self.close()
{{/asyncio}}
{{#asyncio}}async {{/asyncio}}def close(self):
{{#asyncio}}
await self.rest_client.close()
{{/asyncio}}
def close(self):
if self._pool:
self._pool.close()
self._pool.join()
@ -107,6 +112,7 @@ class ApiClient:
atexit.register(self.close)
self._pool = ThreadPool(self.pool_threads)
return self._pool
{{/asyncio}}
@property
def user_agent(self):
@ -230,7 +236,7 @@ class ApiClient:
self.last_response = response_data
return_data = None # assuming derialization is not needed
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)
@ -376,17 +382,19 @@ class ApiClient:
else:
return self.__deserialize_model(data, klass)
def call_api(self, resource_path, method,
{{#asyncio}}async {{/asyncio}}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,
{{^asyncio}}async_req=None, {{/asyncio}}_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.
{{^asyncio}}
To make an async_req request, set the async_req parameter.
{{/asyncio}}
:param resource_path: Path to method endpoint.
:param method: Method to call.
:param path_params: Path parameters in the url.
@ -400,7 +408,9 @@ class ApiClient:
:param response: Response data type.
:param files dict: key -> filename, value -> filepath,
for `multipart/form-data`.
{{^asyncio}}
:param async_req bool: execute request asynchronously
{{/asyncio}}
: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
@ -418,58 +428,69 @@ class ApiClient:
in the spec for a single request.
:type _request_token: dict, optional
:return:
{{#asyncio}}
The response.
{{/asyncio}}
{{^asyncio}}
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.
{{/asyncio}}
"""
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,
)
{{#asyncio}}
return await self.__call_api(*args)
{{/asyncio}}
{{^asyncio}}
if not async_req:
return self.__call_api(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)
return self.__call_api(*args)
return self.pool.apply_async(self.__call_api, (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))
return self.pool.apply_async(self.__call_api, args)
{{/asyncio}}
def request(self, method, url, query_params=None, headers=None,
{{#asyncio}}async {{/asyncio}}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,
return {{#asyncio}}await {{/asyncio}}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,
return {{#asyncio}}await {{/asyncio}}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,
return {{#asyncio}}await {{/asyncio}}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,
return {{#asyncio}}await {{/asyncio}}self.rest_client.post_request(url,
query_params=query_params,
headers=headers,
post_params=post_params,
@ -477,7 +498,7 @@ class ApiClient:
_request_timeout=_request_timeout,
body=body)
elif method == "PUT":
return self.rest_client.put_request(url,
return {{#asyncio}}await {{/asyncio}}self.rest_client.put_request(url,
query_params=query_params,
headers=headers,
post_params=post_params,
@ -485,7 +506,7 @@ class ApiClient:
_request_timeout=_request_timeout,
body=body)
elif method == "PATCH":
return self.rest_client.patch_request(url,
return {{#asyncio}}await {{/asyncio}}self.rest_client.patch_request(url,
query_params=query_params,
headers=headers,
post_params=post_params,
@ -493,7 +514,7 @@ class ApiClient:
_request_timeout=_request_timeout,
body=body)
elif method == "DELETE":
return self.rest_client.delete_request(url,
return {{#asyncio}}await {{/asyncio}}self.rest_client.delete_request(url,
query_params=query_params,
headers=headers,
_preload_content=_preload_content,

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

@ -223,7 +223,7 @@ class ApiClient:
self.last_response = response_data
return_data = None # assuming derialization is not needed
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)
@ -406,27 +406,28 @@ class ApiClient:
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(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)
return self.__call_api(*args)
return self.pool.apply_async(self.__call_api, (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))
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,

34
samples/openapi3/client/petstore/python-aiohttp/petstore_api/api/another_fake_api.py
View File

@ -44,29 +44,14 @@ class AnotherFakeApi:
api_client = ApiClient.get_default()
self.api_client = api_client
@overload
async def call_123_test_special_tags(self, client : Annotated[Client, Field(..., description="client model")], **kwargs) -> Client: # noqa: E501
...
@overload
def call_123_test_special_tags(self, client : Annotated[Client, Field(..., description="client model")], async_req: Optional[bool]=True, **kwargs) -> Client: # noqa: E501
...
@validate_arguments
def call_123_test_special_tags(self, client : Annotated[Client, Field(..., description="client model")], async_req: Optional[bool]=None, **kwargs) -> Union[Client, Awaitable[Client]]: # noqa: E501
async def call_123_test_special_tags(self, client : Annotated[Client, Field(..., description="client model")], **kwargs) -> Client: # noqa: E501
"""To test special tags # noqa: E501
To test special tags and operation ID starting with number # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.call_123_test_special_tags(client, async_req=True)
>>> result = thread.get()
:param client: client model (required)
:type client: Client
: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
@ -80,25 +65,16 @@ class AnotherFakeApi:
if '_preload_content' in kwargs:
message = "Error! Please call the call_123_test_special_tags_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.call_123_test_special_tags_with_http_info(client, **kwargs) # noqa: E501
return await self.call_123_test_special_tags_with_http_info(client, **kwargs) # noqa: E501
@validate_arguments
def call_123_test_special_tags_with_http_info(self, client : Annotated[Client, Field(..., description="client model")], **kwargs) -> ApiResponse: # noqa: E501
async def call_123_test_special_tags_with_http_info(self, client : Annotated[Client, Field(..., description="client model")], **kwargs) -> ApiResponse: # noqa: E501
"""To test special tags # noqa: E501
To test special tags and operation ID starting with number # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.call_123_test_special_tags_with_http_info(client, async_req=True)
>>> result = thread.get()
:param client: client model (required)
:type client: Client
: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.
@ -129,7 +105,6 @@ class AnotherFakeApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -184,7 +159,7 @@ class AnotherFakeApi:
'200': "Client",
}
return self.api_client.call_api(
return await self.api_client.call_api(
'/another-fake/dummy', 'PATCH',
_path_params,
_query_params,
@ -194,7 +169,6 @@ class AnotherFakeApi:
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'),

34
samples/openapi3/client/petstore/python-aiohttp/petstore_api/api/default_api.py
View File

@ -41,26 +41,11 @@ class DefaultApi:
api_client = ApiClient.get_default()
self.api_client = api_client
@overload
async def foo_get(self, **kwargs) -> FooGetDefaultResponse: # noqa: E501
...
@overload
def foo_get(self, async_req: Optional[bool]=True, **kwargs) -> FooGetDefaultResponse: # noqa: E501
...
@validate_arguments
def foo_get(self, async_req: Optional[bool]=None, **kwargs) -> Union[FooGetDefaultResponse, Awaitable[FooGetDefaultResponse]]: # noqa: E501
async def foo_get(self, **kwargs) -> FooGetDefaultResponse: # noqa: E501
"""foo_get # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.foo_get(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
@ -74,22 +59,13 @@ class DefaultApi:
if '_preload_content' in kwargs:
message = "Error! Please call the foo_get_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.foo_get_with_http_info(**kwargs) # noqa: E501
return await self.foo_get_with_http_info(**kwargs) # noqa: E501
@validate_arguments
def foo_get_with_http_info(self, **kwargs) -> ApiResponse: # noqa: E501
async def foo_get_with_http_info(self, **kwargs) -> ApiResponse: # noqa: E501
"""foo_get # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.foo_get_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.
@ -119,7 +95,6 @@ class DefaultApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -163,7 +138,7 @@ class DefaultApi:
_response_types_map = {
}
return self.api_client.call_api(
return await self.api_client.call_api(
'/foo', 'GET',
_path_params,
_query_params,
@ -173,7 +148,6 @@ class DefaultApi:
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'),

714
samples/openapi3/client/petstore/python-aiohttp/petstore_api/api/fake_api.py
View File

File diff suppressed because it is too large Load Diff

34
samples/openapi3/client/petstore/python-aiohttp/petstore_api/api/fake_classname_tags123_api.py
View File

@ -44,29 +44,14 @@ class FakeClassnameTags123Api:
api_client = ApiClient.get_default()
self.api_client = api_client
@overload
async def test_classname(self, client : Annotated[Client, Field(..., description="client model")], **kwargs) -> Client: # noqa: E501
...
@overload
def test_classname(self, client : Annotated[Client, Field(..., description="client model")], async_req: Optional[bool]=True, **kwargs) -> Client: # noqa: E501
...
@validate_arguments
def test_classname(self, client : Annotated[Client, Field(..., description="client model")], async_req: Optional[bool]=None, **kwargs) -> Union[Client, Awaitable[Client]]: # noqa: E501
async def test_classname(self, client : Annotated[Client, Field(..., description="client model")], **kwargs) -> Client: # noqa: E501
"""To test class name in snake case # noqa: E501
To test class name in snake case # 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_classname(client, async_req=True)
>>> result = thread.get()
:param client: client model (required)
:type client: Client
: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
@ -80,25 +65,16 @@ class FakeClassnameTags123Api:
if '_preload_content' in kwargs:
message = "Error! Please call the test_classname_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.test_classname_with_http_info(client, **kwargs) # noqa: E501
return await self.test_classname_with_http_info(client, **kwargs) # noqa: E501
@validate_arguments
def test_classname_with_http_info(self, client : Annotated[Client, Field(..., description="client model")], **kwargs) -> ApiResponse: # noqa: E501
async def test_classname_with_http_info(self, client : Annotated[Client, Field(..., description="client model")], **kwargs) -> ApiResponse: # noqa: E501
"""To test class name in snake case # noqa: E501
To test class name in snake case # 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_classname_with_http_info(client, async_req=True)
>>> result = thread.get()
:param client: client model (required)
:type client: Client
: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.
@ -129,7 +105,6 @@ class FakeClassnameTags123Api:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -184,7 +159,7 @@ class FakeClassnameTags123Api:
'200': "Client",
}
return self.api_client.call_api(
return await self.api_client.call_api(
'/fake_classname_test', 'PATCH',
_path_params,
_query_params,
@ -194,7 +169,6 @@ class FakeClassnameTags123Api:
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'),

306
samples/openapi3/client/petstore/python-aiohttp/petstore_api/api/pet_api.py
View File

@ -47,29 +47,14 @@ class PetApi:
api_client = ApiClient.get_default()
self.api_client = api_client
@overload
async def add_pet(self, pet : Annotated[Pet, Field(..., description="Pet object that needs to be added to the store")], **kwargs) -> None: # noqa: E501
...
@overload
def add_pet(self, pet : Annotated[Pet, Field(..., description="Pet object that needs to be added to the store")], async_req: Optional[bool]=True, **kwargs) -> None: # noqa: E501
...
@validate_arguments
def add_pet(self, pet : Annotated[Pet, Field(..., description="Pet object that needs to be added to the store")], async_req: Optional[bool]=None, **kwargs) -> Union[None, Awaitable[None]]: # noqa: E501
async def add_pet(self, pet : Annotated[Pet, Field(..., description="Pet object that needs to be added to the store")], **kwargs) -> None: # noqa: E501
"""Add a new pet to the store # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.add_pet(pet, async_req=True)
>>> result = thread.get()
:param pet: Pet object that needs to be added to the store (required)
:type pet: Pet
: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
@ -83,25 +68,16 @@ class PetApi:
if '_preload_content' in kwargs:
message = "Error! Please call the add_pet_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.add_pet_with_http_info(pet, **kwargs) # noqa: E501
return await self.add_pet_with_http_info(pet, **kwargs) # noqa: E501
@validate_arguments
def add_pet_with_http_info(self, pet : Annotated[Pet, Field(..., description="Pet object that needs to be added to the store")], **kwargs) -> ApiResponse: # noqa: E501
async def add_pet_with_http_info(self, pet : Annotated[Pet, Field(..., description="Pet object that needs to be added to the store")], **kwargs) -> ApiResponse: # noqa: E501
"""Add a new pet to the store # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.add_pet_with_http_info(pet, async_req=True)
>>> result = thread.get()
:param pet: Pet object that needs to be added to the store (required)
:type pet: Pet
: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.
@ -132,7 +108,6 @@ class PetApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -181,7 +156,7 @@ class PetApi:
_response_types_map = {}
return self.api_client.call_api(
return await self.api_client.call_api(
'/pet', 'POST',
_path_params,
_query_params,
@ -191,38 +166,22 @@ class PetApi:
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'))
@overload
async def delete_pet(self, pet_id : Annotated[StrictInt, Field(..., description="Pet id to delete")], api_key : Optional[StrictStr] = None, **kwargs) -> None: # noqa: E501
...
@overload
def delete_pet(self, pet_id : Annotated[StrictInt, Field(..., description="Pet id to delete")], api_key : Optional[StrictStr] = None, async_req: Optional[bool]=True, **kwargs) -> None: # noqa: E501
...
@validate_arguments
def delete_pet(self, pet_id : Annotated[StrictInt, Field(..., description="Pet id to delete")], api_key : Optional[StrictStr] = None, async_req: Optional[bool]=None, **kwargs) -> Union[None, Awaitable[None]]: # noqa: E501
async def delete_pet(self, pet_id : Annotated[StrictInt, Field(..., description="Pet id to delete")], api_key : Optional[StrictStr] = None, **kwargs) -> None: # noqa: E501
"""Deletes a pet # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.delete_pet(pet_id, api_key, async_req=True)
>>> result = thread.get()
:param pet_id: Pet id to delete (required)
:type pet_id: int
:param api_key:
:type api_key: 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
@ -236,27 +195,18 @@ class PetApi:
if '_preload_content' in kwargs:
message = "Error! Please call the delete_pet_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.delete_pet_with_http_info(pet_id, api_key, **kwargs) # noqa: E501
return await self.delete_pet_with_http_info(pet_id, api_key, **kwargs) # noqa: E501
@validate_arguments
def delete_pet_with_http_info(self, pet_id : Annotated[StrictInt, Field(..., description="Pet id to delete")], api_key : Optional[StrictStr] = None, **kwargs) -> ApiResponse: # noqa: E501
async def delete_pet_with_http_info(self, pet_id : Annotated[StrictInt, Field(..., description="Pet id to delete")], api_key : Optional[StrictStr] = None, **kwargs) -> ApiResponse: # noqa: E501
"""Deletes a pet # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.delete_pet_with_http_info(pet_id, api_key, async_req=True)
>>> result = thread.get()
:param pet_id: Pet id to delete (required)
:type pet_id: int
:param api_key:
:type api_key: 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.
@ -288,7 +238,6 @@ class PetApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -333,7 +282,7 @@ class PetApi:
_response_types_map = {}
return self.api_client.call_api(
return await self.api_client.call_api(
'/pet/{petId}', 'DELETE',
_path_params,
_query_params,
@ -343,36 +292,20 @@ class PetApi:
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'))
@overload
async def find_pets_by_status(self, status : Annotated[conlist(StrictStr), Field(..., description="Status values that need to be considered for filter")], **kwargs) -> List[Pet]: # noqa: E501
...
@overload
def find_pets_by_status(self, status : Annotated[conlist(StrictStr), Field(..., description="Status values that need to be considered for filter")], async_req: Optional[bool]=True, **kwargs) -> List[Pet]: # noqa: E501
...
@validate_arguments
def find_pets_by_status(self, status : Annotated[conlist(StrictStr), Field(..., description="Status values that need to be considered for filter")], async_req: Optional[bool]=None, **kwargs) -> Union[List[Pet], Awaitable[List[Pet]]]: # noqa: E501
async def find_pets_by_status(self, status : Annotated[conlist(StrictStr), Field(..., description="Status values that need to be considered for filter")], **kwargs) -> List[Pet]: # noqa: E501
"""Finds Pets by status # noqa: E501
Multiple status values can be provided with comma separated strings # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.find_pets_by_status(status, async_req=True)
>>> result = thread.get()
:param status: Status values that need to be considered for filter (required)
:type status: List[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
@ -386,25 +319,16 @@ class PetApi:
if '_preload_content' in kwargs:
message = "Error! Please call the find_pets_by_status_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.find_pets_by_status_with_http_info(status, **kwargs) # noqa: E501
return await self.find_pets_by_status_with_http_info(status, **kwargs) # noqa: E501
@validate_arguments
def find_pets_by_status_with_http_info(self, status : Annotated[conlist(StrictStr), Field(..., description="Status values that need to be considered for filter")], **kwargs) -> ApiResponse: # noqa: E501
async def find_pets_by_status_with_http_info(self, status : Annotated[conlist(StrictStr), Field(..., description="Status values that need to be considered for filter")], **kwargs) -> ApiResponse: # noqa: E501
"""Finds Pets by status # noqa: E501
Multiple status values can be provided with comma separated strings # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.find_pets_by_status_with_http_info(status, async_req=True)
>>> result = thread.get()
:param status: Status values that need to be considered for filter (required)
:type status: List[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.
@ -435,7 +359,6 @@ class PetApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -485,7 +408,7 @@ class PetApi:
'400': None,
}
return self.api_client.call_api(
return await self.api_client.call_api(
'/pet/findByStatus', 'GET',
_path_params,
_query_params,
@ -495,36 +418,20 @@ class PetApi:
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'))
@overload
async def find_pets_by_tags(self, tags : Annotated[conlist(StrictStr, unique_items=True), Field(..., description="Tags to filter by")], **kwargs) -> List[Pet]: # noqa: E501
...
@overload
def find_pets_by_tags(self, tags : Annotated[conlist(StrictStr, unique_items=True), Field(..., description="Tags to filter by")], async_req: Optional[bool]=True, **kwargs) -> List[Pet]: # noqa: E501
...
@validate_arguments
def find_pets_by_tags(self, tags : Annotated[conlist(StrictStr, unique_items=True), Field(..., description="Tags to filter by")], async_req: Optional[bool]=None, **kwargs) -> Union[List[Pet], Awaitable[List[Pet]]]: # noqa: E501
async def find_pets_by_tags(self, tags : Annotated[conlist(StrictStr, unique_items=True), Field(..., description="Tags to filter by")], **kwargs) -> List[Pet]: # noqa: E501
"""(Deprecated) Finds Pets by tags # noqa: E501
Multiple tags can be provided with comma separated strings. Use tag1, tag2, tag3 for testing. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.find_pets_by_tags(tags, async_req=True)
>>> result = thread.get()
:param tags: Tags to filter by (required)
:type tags: List[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
@ -538,25 +445,16 @@ class PetApi:
if '_preload_content' in kwargs:
message = "Error! Please call the find_pets_by_tags_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.find_pets_by_tags_with_http_info(tags, **kwargs) # noqa: E501
return await self.find_pets_by_tags_with_http_info(tags, **kwargs) # noqa: E501
@validate_arguments
def find_pets_by_tags_with_http_info(self, tags : Annotated[conlist(StrictStr, unique_items=True), Field(..., description="Tags to filter by")], **kwargs) -> ApiResponse: # noqa: E501
async def find_pets_by_tags_with_http_info(self, tags : Annotated[conlist(StrictStr, unique_items=True), Field(..., description="Tags to filter by")], **kwargs) -> ApiResponse: # noqa: E501
"""(Deprecated) Finds Pets by tags # noqa: E501
Multiple tags can be provided with comma separated strings. Use tag1, tag2, tag3 for testing. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.find_pets_by_tags_with_http_info(tags, async_req=True)
>>> result = thread.get()
:param tags: Tags to filter by (required)
:type tags: List[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.
@ -589,7 +487,6 @@ class PetApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -639,7 +536,7 @@ class PetApi:
'400': None,
}
return self.api_client.call_api(
return await self.api_client.call_api(
'/pet/findByTags', 'GET',
_path_params,
_query_params,
@ -649,36 +546,20 @@ class PetApi:
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'))
@overload
async def get_pet_by_id(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to return")], **kwargs) -> Pet: # noqa: E501
...
@overload
def get_pet_by_id(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to return")], async_req: Optional[bool]=True, **kwargs) -> Pet: # noqa: E501
...
@validate_arguments
def get_pet_by_id(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to return")], async_req: Optional[bool]=None, **kwargs) -> Union[Pet, Awaitable[Pet]]: # noqa: E501
async def get_pet_by_id(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to return")], **kwargs) -> Pet: # noqa: E501
"""Find pet by ID # noqa: E501
Returns a single pet # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.get_pet_by_id(pet_id, async_req=True)
>>> result = thread.get()
:param pet_id: ID of pet to return (required)
:type pet_id: int
: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
@ -692,25 +573,16 @@ class PetApi:
if '_preload_content' in kwargs:
message = "Error! Please call the get_pet_by_id_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.get_pet_by_id_with_http_info(pet_id, **kwargs) # noqa: E501
return await self.get_pet_by_id_with_http_info(pet_id, **kwargs) # noqa: E501
@validate_arguments
def get_pet_by_id_with_http_info(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to return")], **kwargs) -> ApiResponse: # noqa: E501
async def get_pet_by_id_with_http_info(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to return")], **kwargs) -> ApiResponse: # noqa: E501
"""Find pet by ID # noqa: E501
Returns a single pet # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.get_pet_by_id_with_http_info(pet_id, async_req=True)
>>> result = thread.get()
:param pet_id: ID of pet to return (required)
:type pet_id: int
: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.
@ -741,7 +613,6 @@ class PetApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -791,7 +662,7 @@ class PetApi:
'404': None,
}
return self.api_client.call_api(
return await self.api_client.call_api(
'/pet/{petId}', 'GET',
_path_params,
_query_params,
@ -801,36 +672,20 @@ class PetApi:
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'))
@overload
async def update_pet(self, pet : Annotated[Pet, Field(..., description="Pet object that needs to be added to the store")], **kwargs) -> None: # noqa: E501
...
@overload
def update_pet(self, pet : Annotated[Pet, Field(..., description="Pet object that needs to be added to the store")], async_req: Optional[bool]=True, **kwargs) -> None: # noqa: E501
...
@validate_arguments
def update_pet(self, pet : Annotated[Pet, Field(..., description="Pet object that needs to be added to the store")], async_req: Optional[bool]=None, **kwargs) -> Union[None, Awaitable[None]]: # noqa: E501
async def update_pet(self, pet : Annotated[Pet, Field(..., description="Pet object that needs to be added to the store")], **kwargs) -> None: # noqa: E501
"""Update an existing pet # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.update_pet(pet, async_req=True)
>>> result = thread.get()
:param pet: Pet object that needs to be added to the store (required)
:type pet: Pet
: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
@ -844,25 +699,16 @@ class PetApi:
if '_preload_content' in kwargs:
message = "Error! Please call the update_pet_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.update_pet_with_http_info(pet, **kwargs) # noqa: E501
return await self.update_pet_with_http_info(pet, **kwargs) # noqa: E501
@validate_arguments
def update_pet_with_http_info(self, pet : Annotated[Pet, Field(..., description="Pet object that needs to be added to the store")], **kwargs) -> ApiResponse: # noqa: E501
async def update_pet_with_http_info(self, pet : Annotated[Pet, Field(..., description="Pet object that needs to be added to the store")], **kwargs) -> ApiResponse: # noqa: E501
"""Update an existing pet # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.update_pet_with_http_info(pet, async_req=True)
>>> result = thread.get()
:param pet: Pet object that needs to be added to the store (required)
:type pet: Pet
: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.
@ -893,7 +739,6 @@ class PetApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -942,7 +787,7 @@ class PetApi:
_response_types_map = {}
return self.api_client.call_api(
return await self.api_client.call_api(
'/pet', 'PUT',
_path_params,
_query_params,
@ -952,31 +797,17 @@ class PetApi:
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'))
@overload
async def update_pet_with_form(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet that needs to be updated")], name : Annotated[Optional[StrictStr], Field(description="Updated name of the pet")] = None, status : Annotated[Optional[StrictStr], Field(description="Updated status of the pet")] = None, **kwargs) -> None: # noqa: E501
...
@overload
def update_pet_with_form(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet that needs to be updated")], name : Annotated[Optional[StrictStr], Field(description="Updated name of the pet")] = None, status : Annotated[Optional[StrictStr], Field(description="Updated status of the pet")] = None, async_req: Optional[bool]=True, **kwargs) -> None: # noqa: E501
...
@validate_arguments
def update_pet_with_form(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet that needs to be updated")], name : Annotated[Optional[StrictStr], Field(description="Updated name of the pet")] = None, status : Annotated[Optional[StrictStr], Field(description="Updated status of the pet")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[None, Awaitable[None]]: # noqa: E501
async def update_pet_with_form(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet that needs to be updated")], name : Annotated[Optional[StrictStr], Field(description="Updated name of the pet")] = None, status : Annotated[Optional[StrictStr], Field(description="Updated status of the pet")] = None, **kwargs) -> None: # noqa: E501
"""Updates a pet in the store with form data # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.update_pet_with_form(pet_id, name, status, async_req=True)
>>> result = thread.get()
:param pet_id: ID of pet that needs to be updated (required)
:type pet_id: int
@ -984,8 +815,6 @@ class PetApi:
:type name: str
:param status: Updated status of the pet
:type status: 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
@ -999,20 +828,13 @@ class PetApi:
if '_preload_content' in kwargs:
message = "Error! Please call the update_pet_with_form_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.update_pet_with_form_with_http_info(pet_id, name, status, **kwargs) # noqa: E501
return await self.update_pet_with_form_with_http_info(pet_id, name, status, **kwargs) # noqa: E501
@validate_arguments
def update_pet_with_form_with_http_info(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet that needs to be updated")], name : Annotated[Optional[StrictStr], Field(description="Updated name of the pet")] = None, status : Annotated[Optional[StrictStr], Field(description="Updated status of the pet")] = None, **kwargs) -> ApiResponse: # noqa: E501
async def update_pet_with_form_with_http_info(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet that needs to be updated")], name : Annotated[Optional[StrictStr], Field(description="Updated name of the pet")] = None, status : Annotated[Optional[StrictStr], Field(description="Updated status of the pet")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""Updates a pet in the store with form data # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.update_pet_with_form_with_http_info(pet_id, name, status, async_req=True)
>>> result = thread.get()
:param pet_id: ID of pet that needs to be updated (required)
:type pet_id: int
@ -1020,8 +842,6 @@ class PetApi:
:type name: str
:param status: Updated status of the pet
:type status: 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.
@ -1054,7 +874,6 @@ class PetApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -1109,7 +928,7 @@ class PetApi:
_response_types_map = {}
return self.api_client.call_api(
return await self.api_client.call_api(
'/pet/{petId}', 'POST',
_path_params,
_query_params,
@ -1119,31 +938,17 @@ class PetApi:
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'))
@overload
async def upload_file(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to update")], additional_metadata : Annotated[Optional[StrictStr], Field(description="Additional data to pass to server")] = None, file : Annotated[Optional[Union[StrictBytes, StrictStr]], Field(description="file to upload")] = None, **kwargs) -> ApiResponse: # noqa: E501
...
@overload
def upload_file(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to update")], additional_metadata : Annotated[Optional[StrictStr], Field(description="Additional data to pass to server")] = None, file : Annotated[Optional[Union[StrictBytes, StrictStr]], Field(description="file to upload")] = None, async_req: Optional[bool]=True, **kwargs) -> ApiResponse: # noqa: E501
...
@validate_arguments
def upload_file(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to update")], additional_metadata : Annotated[Optional[StrictStr], Field(description="Additional data to pass to server")] = None, file : Annotated[Optional[Union[StrictBytes, StrictStr]], Field(description="file to upload")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[ApiResponse, Awaitable[ApiResponse]]: # noqa: E501
async def upload_file(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to update")], additional_metadata : Annotated[Optional[StrictStr], Field(description="Additional data to pass to server")] = None, file : Annotated[Optional[Union[StrictBytes, StrictStr]], Field(description="file to upload")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""uploads an image # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.upload_file(pet_id, additional_metadata, file, async_req=True)
>>> result = thread.get()
:param pet_id: ID of pet to update (required)
:type pet_id: int
@ -1151,8 +956,6 @@ class PetApi:
:type additional_metadata: str
:param file: file to upload
:type file: bytearray
: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
@ -1166,20 +969,13 @@ class PetApi:
if '_preload_content' in kwargs:
message = "Error! Please call the upload_file_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.upload_file_with_http_info(pet_id, additional_metadata, file, **kwargs) # noqa: E501
return await self.upload_file_with_http_info(pet_id, additional_metadata, file, **kwargs) # noqa: E501
@validate_arguments
def upload_file_with_http_info(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to update")], additional_metadata : Annotated[Optional[StrictStr], Field(description="Additional data to pass to server")] = None, file : Annotated[Optional[Union[StrictBytes, StrictStr]], Field(description="file to upload")] = None, **kwargs) -> ApiResponse: # noqa: E501
async def upload_file_with_http_info(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to update")], additional_metadata : Annotated[Optional[StrictStr], Field(description="Additional data to pass to server")] = None, file : Annotated[Optional[Union[StrictBytes, StrictStr]], Field(description="file to upload")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""uploads an image # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.upload_file_with_http_info(pet_id, additional_metadata, file, async_req=True)
>>> result = thread.get()
:param pet_id: ID of pet to update (required)
:type pet_id: int
@ -1187,8 +983,6 @@ class PetApi:
:type additional_metadata: str
:param file: file to upload
:type file: bytearray
: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.
@ -1221,7 +1015,6 @@ class PetApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -1282,7 +1075,7 @@ class PetApi:
'200': "ApiResponse",
}
return self.api_client.call_api(
return await self.api_client.call_api(
'/pet/{petId}/uploadImage', 'POST',
_path_params,
_query_params,
@ -1292,31 +1085,17 @@ class PetApi:
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'))
@overload
async def upload_file_with_required_file(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to update")], required_file : Annotated[Union[StrictBytes, StrictStr], Field(..., description="file to upload")], additional_metadata : Annotated[Optional[StrictStr], Field(description="Additional data to pass to server")] = None, **kwargs) -> ApiResponse: # noqa: E501
...
@overload
def upload_file_with_required_file(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to update")], required_file : Annotated[Union[StrictBytes, StrictStr], Field(..., description="file to upload")], additional_metadata : Annotated[Optional[StrictStr], Field(description="Additional data to pass to server")] = None, async_req: Optional[bool]=True, **kwargs) -> ApiResponse: # noqa: E501
...
@validate_arguments
def upload_file_with_required_file(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to update")], required_file : Annotated[Union[StrictBytes, StrictStr], Field(..., description="file to upload")], additional_metadata : Annotated[Optional[StrictStr], Field(description="Additional data to pass to server")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[ApiResponse, Awaitable[ApiResponse]]: # noqa: E501
async def upload_file_with_required_file(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to update")], required_file : Annotated[Union[StrictBytes, StrictStr], Field(..., description="file to upload")], additional_metadata : Annotated[Optional[StrictStr], Field(description="Additional data to pass to server")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""uploads an image (required) # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.upload_file_with_required_file(pet_id, required_file, additional_metadata, async_req=True)
>>> result = thread.get()
:param pet_id: ID of pet to update (required)
:type pet_id: int
@ -1324,8 +1103,6 @@ class PetApi:
:type required_file: bytearray
:param additional_metadata: Additional data to pass to server
:type additional_metadata: 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
@ -1339,20 +1116,13 @@ class PetApi:
if '_preload_content' in kwargs:
message = "Error! Please call the upload_file_with_required_file_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.upload_file_with_required_file_with_http_info(pet_id, required_file, additional_metadata, **kwargs) # noqa: E501
return await self.upload_file_with_required_file_with_http_info(pet_id, required_file, additional_metadata, **kwargs) # noqa: E501
@validate_arguments
def upload_file_with_required_file_with_http_info(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to update")], required_file : Annotated[Union[StrictBytes, StrictStr], Field(..., description="file to upload")], additional_metadata : Annotated[Optional[StrictStr], Field(description="Additional data to pass to server")] = None, **kwargs) -> ApiResponse: # noqa: E501
async def upload_file_with_required_file_with_http_info(self, pet_id : Annotated[StrictInt, Field(..., description="ID of pet to update")], required_file : Annotated[Union[StrictBytes, StrictStr], Field(..., description="file to upload")], additional_metadata : Annotated[Optional[StrictStr], Field(description="Additional data to pass to server")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""uploads an image (required) # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.upload_file_with_required_file_with_http_info(pet_id, required_file, additional_metadata, async_req=True)
>>> result = thread.get()
:param pet_id: ID of pet to update (required)
:type pet_id: int
@ -1360,8 +1130,6 @@ class PetApi:
:type required_file: bytearray
:param additional_metadata: Additional data to pass to server
:type additional_metadata: 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.
@ -1394,7 +1162,6 @@ class PetApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -1455,7 +1222,7 @@ class PetApi:
'200': "ApiResponse",
}
return self.api_client.call_api(
return await self.api_client.call_api(
'/fake/{petId}/uploadImageWithRequiredFile', 'POST',
_path_params,
_query_params,
@ -1465,7 +1232,6 @@ class PetApi:
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'),

136
samples/openapi3/client/petstore/python-aiohttp/petstore_api/api/store_api.py
View File

@ -46,29 +46,14 @@ class StoreApi:
api_client = ApiClient.get_default()
self.api_client = api_client
@overload
async def delete_order(self, order_id : Annotated[StrictStr, Field(..., description="ID of the order that needs to be deleted")], **kwargs) -> None: # noqa: E501
...
@overload
def delete_order(self, order_id : Annotated[StrictStr, Field(..., description="ID of the order that needs to be deleted")], async_req: Optional[bool]=True, **kwargs) -> None: # noqa: E501
...
@validate_arguments
def delete_order(self, order_id : Annotated[StrictStr, Field(..., description="ID of the order that needs to be deleted")], async_req: Optional[bool]=None, **kwargs) -> Union[None, Awaitable[None]]: # noqa: E501
async def delete_order(self, order_id : Annotated[StrictStr, Field(..., description="ID of the order that needs to be deleted")], **kwargs) -> None: # noqa: E501
"""Delete purchase order by ID # noqa: E501
For valid response try integer IDs with value < 1000. Anything above 1000 or nonintegers will generate API errors # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.delete_order(order_id, async_req=True)
>>> result = thread.get()
:param order_id: ID of the order that needs to be deleted (required)
:type order_id: 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
@ -82,25 +67,16 @@ class StoreApi:
if '_preload_content' in kwargs:
message = "Error! Please call the delete_order_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.delete_order_with_http_info(order_id, **kwargs) # noqa: E501
return await self.delete_order_with_http_info(order_id, **kwargs) # noqa: E501
@validate_arguments
def delete_order_with_http_info(self, order_id : Annotated[StrictStr, Field(..., description="ID of the order that needs to be deleted")], **kwargs) -> ApiResponse: # noqa: E501
async def delete_order_with_http_info(self, order_id : Annotated[StrictStr, Field(..., description="ID of the order that needs to be deleted")], **kwargs) -> ApiResponse: # noqa: E501
"""Delete purchase order by ID # noqa: E501
For valid response try integer IDs with value < 1000. Anything above 1000 or nonintegers will generate API errors # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.delete_order_with_http_info(order_id, async_req=True)
>>> result = thread.get()
:param order_id: ID of the order that needs to be deleted (required)
:type order_id: 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.
@ -131,7 +107,6 @@ class StoreApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -173,7 +148,7 @@ class StoreApi:
_response_types_map = {}
return self.api_client.call_api(
return await self.api_client.call_api(
'/store/order/{order_id}', 'DELETE',
_path_params,
_query_params,
@ -183,34 +158,18 @@ class StoreApi:
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'))
@overload
async def get_inventory(self, **kwargs) -> Dict[str, int]: # noqa: E501
...
@overload
def get_inventory(self, async_req: Optional[bool]=True, **kwargs) -> Dict[str, int]: # noqa: E501
...
@validate_arguments
def get_inventory(self, async_req: Optional[bool]=None, **kwargs) -> Union[Dict[str, int], Awaitable[Dict[str, int]]]: # noqa: E501
async def get_inventory(self, **kwargs) -> Dict[str, int]: # noqa: E501
"""Returns pet inventories by status # noqa: E501
Returns a map of status codes to quantities # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.get_inventory(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
@ -224,23 +183,14 @@ class StoreApi:
if '_preload_content' in kwargs:
message = "Error! Please call the get_inventory_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.get_inventory_with_http_info(**kwargs) # noqa: E501
return await self.get_inventory_with_http_info(**kwargs) # noqa: E501
@validate_arguments
def get_inventory_with_http_info(self, **kwargs) -> ApiResponse: # noqa: E501
async def get_inventory_with_http_info(self, **kwargs) -> ApiResponse: # noqa: E501
"""Returns pet inventories by status # noqa: E501
Returns a map of status codes to quantities # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.get_inventory_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.
@ -270,7 +220,6 @@ class StoreApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -315,7 +264,7 @@ class StoreApi:
'200': "Dict[str, int]",
}
return self.api_client.call_api(
return await self.api_client.call_api(
'/store/inventory', 'GET',
_path_params,
_query_params,
@ -325,36 +274,20 @@ class StoreApi:
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'))
@overload
async def get_order_by_id(self, order_id : Annotated[conint(strict=True, le=5, ge=1), Field(..., description="ID of pet that needs to be fetched")], **kwargs) -> Order: # noqa: E501
...
@overload
def get_order_by_id(self, order_id : Annotated[conint(strict=True, le=5, ge=1), Field(..., description="ID of pet that needs to be fetched")], async_req: Optional[bool]=True, **kwargs) -> Order: # noqa: E501
...
@validate_arguments
def get_order_by_id(self, order_id : Annotated[conint(strict=True, le=5, ge=1), Field(..., description="ID of pet that needs to be fetched")], async_req: Optional[bool]=None, **kwargs) -> Union[Order, Awaitable[Order]]: # noqa: E501
async def get_order_by_id(self, order_id : Annotated[conint(strict=True, le=5, ge=1), Field(..., description="ID of pet that needs to be fetched")], **kwargs) -> Order: # noqa: E501
"""Find purchase order by ID # noqa: E501
For valid response try integer IDs with value <= 5 or > 10. Other values will generated exceptions # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.get_order_by_id(order_id, async_req=True)
>>> result = thread.get()
:param order_id: ID of pet that needs to be fetched (required)
:type order_id: int
: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
@ -368,25 +301,16 @@ class StoreApi:
if '_preload_content' in kwargs:
message = "Error! Please call the get_order_by_id_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.get_order_by_id_with_http_info(order_id, **kwargs) # noqa: E501
return await self.get_order_by_id_with_http_info(order_id, **kwargs) # noqa: E501
@validate_arguments
def get_order_by_id_with_http_info(self, order_id : Annotated[conint(strict=True, le=5, ge=1), Field(..., description="ID of pet that needs to be fetched")], **kwargs) -> ApiResponse: # noqa: E501
async def get_order_by_id_with_http_info(self, order_id : Annotated[conint(strict=True, le=5, ge=1), Field(..., description="ID of pet that needs to be fetched")], **kwargs) -> ApiResponse: # noqa: E501
"""Find purchase order by ID # noqa: E501
For valid response try integer IDs with value <= 5 or > 10. Other values will generated exceptions # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.get_order_by_id_with_http_info(order_id, async_req=True)
>>> result = thread.get()
:param order_id: ID of pet that needs to be fetched (required)
:type order_id: int
: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.
@ -417,7 +341,6 @@ class StoreApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -467,7 +390,7 @@ class StoreApi:
'404': None,
}
return self.api_client.call_api(
return await self.api_client.call_api(
'/store/order/{order_id}', 'GET',
_path_params,
_query_params,
@ -477,36 +400,20 @@ class StoreApi:
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'))
@overload
async def place_order(self, order : Annotated[Order, Field(..., description="order placed for purchasing the pet")], **kwargs) -> Order: # noqa: E501
...
@overload
def place_order(self, order : Annotated[Order, Field(..., description="order placed for purchasing the pet")], async_req: Optional[bool]=True, **kwargs) -> Order: # noqa: E501
...
@validate_arguments
def place_order(self, order : Annotated[Order, Field(..., description="order placed for purchasing the pet")], async_req: Optional[bool]=None, **kwargs) -> Union[Order, Awaitable[Order]]: # noqa: E501
async def place_order(self, order : Annotated[Order, Field(..., description="order placed for purchasing the pet")], **kwargs) -> Order: # noqa: E501
"""Place an order for a pet # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.place_order(order, async_req=True)
>>> result = thread.get()
:param order: order placed for purchasing the pet (required)
:type order: Order
: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
@ -520,25 +427,16 @@ class StoreApi:
if '_preload_content' in kwargs:
message = "Error! Please call the place_order_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.place_order_with_http_info(order, **kwargs) # noqa: E501
return await self.place_order_with_http_info(order, **kwargs) # noqa: E501
@validate_arguments
def place_order_with_http_info(self, order : Annotated[Order, Field(..., description="order placed for purchasing the pet")], **kwargs) -> ApiResponse: # noqa: E501
async def place_order_with_http_info(self, order : Annotated[Order, Field(..., description="order placed for purchasing the pet")], **kwargs) -> ApiResponse: # noqa: E501
"""Place an order for a pet # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.place_order_with_http_info(order, async_req=True)
>>> result = thread.get()
:param order: order placed for purchasing the pet (required)
:type order: Order
: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.
@ -569,7 +467,6 @@ class StoreApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -625,7 +522,7 @@ class StoreApi:
'400': None,
}
return self.api_client.call_api(
return await self.api_client.call_api(
'/store/order', 'POST',
_path_params,
_query_params,
@ -635,7 +532,6 @@ class StoreApi:
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'),

272
samples/openapi3/client/petstore/python-aiohttp/petstore_api/api/user_api.py
View File

@ -44,29 +44,14 @@ class UserApi:
api_client = ApiClient.get_default()
self.api_client = api_client
@overload
async def create_user(self, user : Annotated[User, Field(..., description="Created user object")], **kwargs) -> None: # noqa: E501
...
@overload
def create_user(self, user : Annotated[User, Field(..., description="Created user object")], async_req: Optional[bool]=True, **kwargs) -> None: # noqa: E501
...
@validate_arguments
def create_user(self, user : Annotated[User, Field(..., description="Created user object")], async_req: Optional[bool]=None, **kwargs) -> Union[None, Awaitable[None]]: # noqa: E501
async def create_user(self, user : Annotated[User, Field(..., description="Created user object")], **kwargs) -> None: # noqa: E501
"""Create user # noqa: E501
This can only be done by the logged in user. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.create_user(user, async_req=True)
>>> result = thread.get()
:param user: Created user object (required)
:type user: User
: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
@ -80,25 +65,16 @@ class UserApi:
if '_preload_content' in kwargs:
message = "Error! Please call the create_user_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.create_user_with_http_info(user, **kwargs) # noqa: E501
return await self.create_user_with_http_info(user, **kwargs) # noqa: E501
@validate_arguments
def create_user_with_http_info(self, user : Annotated[User, Field(..., description="Created user object")], **kwargs) -> ApiResponse: # noqa: E501
async def create_user_with_http_info(self, user : Annotated[User, Field(..., description="Created user object")], **kwargs) -> ApiResponse: # noqa: E501
"""Create user # noqa: E501
This can only be done by the logged in user. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.create_user_with_http_info(user, async_req=True)
>>> result = thread.get()
:param user: Created user object (required)
:type user: User
: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.
@ -143,7 +119,6 @@ class UserApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -192,7 +167,7 @@ class UserApi:
_response_types_map = {}
return self.api_client.call_api(
return await self.api_client.call_api(
'/user', 'POST',
_path_params,
_query_params,
@ -202,7 +177,6 @@ class UserApi:
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'),
@ -210,29 +184,14 @@ class UserApi:
collection_formats=_collection_formats,
_request_auth=_params.get('_request_auth'))
@overload
async def create_users_with_array_input(self, user : Annotated[conlist(User), Field(..., description="List of user object")], **kwargs) -> None: # noqa: E501
...
@overload
def create_users_with_array_input(self, user : Annotated[conlist(User), Field(..., description="List of user object")], async_req: Optional[bool]=True, **kwargs) -> None: # noqa: E501
...
@validate_arguments
def create_users_with_array_input(self, user : Annotated[conlist(User), Field(..., description="List of user object")], async_req: Optional[bool]=None, **kwargs) -> Union[None, Awaitable[None]]: # noqa: E501
async def create_users_with_array_input(self, user : Annotated[conlist(User), Field(..., description="List of user object")], **kwargs) -> None: # noqa: E501
"""Creates list of users with given input array # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.create_users_with_array_input(user, async_req=True)
>>> result = thread.get()
:param user: List of user object (required)
:type user: List[User]
: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
@ -246,25 +205,16 @@ class UserApi:
if '_preload_content' in kwargs:
message = "Error! Please call the create_users_with_array_input_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.create_users_with_array_input_with_http_info(user, **kwargs) # noqa: E501
return await self.create_users_with_array_input_with_http_info(user, **kwargs) # noqa: E501
@validate_arguments
def create_users_with_array_input_with_http_info(self, user : Annotated[conlist(User), Field(..., description="List of user object")], **kwargs) -> ApiResponse: # noqa: E501
async def create_users_with_array_input_with_http_info(self, user : Annotated[conlist(User), Field(..., description="List of user object")], **kwargs) -> ApiResponse: # noqa: E501
"""Creates list of users with given input array # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.create_users_with_array_input_with_http_info(user, async_req=True)
>>> result = thread.get()
:param user: List of user object (required)
:type user: List[User]
: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.
@ -295,7 +245,6 @@ class UserApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -344,7 +293,7 @@ class UserApi:
_response_types_map = {}
return self.api_client.call_api(
return await self.api_client.call_api(
'/user/createWithArray', 'POST',
_path_params,
_query_params,
@ -354,36 +303,20 @@ class UserApi:
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'))
@overload
async def create_users_with_list_input(self, user : Annotated[conlist(User), Field(..., description="List of user object")], **kwargs) -> None: # noqa: E501
...
@overload
def create_users_with_list_input(self, user : Annotated[conlist(User), Field(..., description="List of user object")], async_req: Optional[bool]=True, **kwargs) -> None: # noqa: E501
...
@validate_arguments
def create_users_with_list_input(self, user : Annotated[conlist(User), Field(..., description="List of user object")], async_req: Optional[bool]=None, **kwargs) -> Union[None, Awaitable[None]]: # noqa: E501
async def create_users_with_list_input(self, user : Annotated[conlist(User), Field(..., description="List of user object")], **kwargs) -> None: # noqa: E501
"""Creates list of users with given input array # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.create_users_with_list_input(user, async_req=True)
>>> result = thread.get()
:param user: List of user object (required)
:type user: List[User]
: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
@ -397,25 +330,16 @@ class UserApi:
if '_preload_content' in kwargs:
message = "Error! Please call the create_users_with_list_input_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.create_users_with_list_input_with_http_info(user, **kwargs) # noqa: E501
return await self.create_users_with_list_input_with_http_info(user, **kwargs) # noqa: E501
@validate_arguments
def create_users_with_list_input_with_http_info(self, user : Annotated[conlist(User), Field(..., description="List of user object")], **kwargs) -> ApiResponse: # noqa: E501
async def create_users_with_list_input_with_http_info(self, user : Annotated[conlist(User), Field(..., description="List of user object")], **kwargs) -> ApiResponse: # noqa: E501
"""Creates list of users with given input array # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.create_users_with_list_input_with_http_info(user, async_req=True)
>>> result = thread.get()
:param user: List of user object (required)
:type user: List[User]
: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.
@ -446,7 +370,6 @@ class UserApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -495,7 +418,7 @@ class UserApi:
_response_types_map = {}
return self.api_client.call_api(
return await self.api_client.call_api(
'/user/createWithList', 'POST',
_path_params,
_query_params,
@ -505,36 +428,20 @@ class UserApi:
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'))
@overload
async def delete_user(self, username : Annotated[StrictStr, Field(..., description="The name that needs to be deleted")], **kwargs) -> None: # noqa: E501
...
@overload
def delete_user(self, username : Annotated[StrictStr, Field(..., description="The name that needs to be deleted")], async_req: Optional[bool]=True, **kwargs) -> None: # noqa: E501
...
@validate_arguments
def delete_user(self, username : Annotated[StrictStr, Field(..., description="The name that needs to be deleted")], async_req: Optional[bool]=None, **kwargs) -> Union[None, Awaitable[None]]: # noqa: E501
async def delete_user(self, username : Annotated[StrictStr, Field(..., description="The name that needs to be deleted")], **kwargs) -> None: # noqa: E501
"""Delete user # noqa: E501
This can only be done by the logged in user. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.delete_user(username, async_req=True)
>>> result = thread.get()
:param username: The name that needs to be deleted (required)
:type username: 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
@ -548,25 +455,16 @@ class UserApi:
if '_preload_content' in kwargs:
message = "Error! Please call the delete_user_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.delete_user_with_http_info(username, **kwargs) # noqa: E501
return await self.delete_user_with_http_info(username, **kwargs) # noqa: E501
@validate_arguments
def delete_user_with_http_info(self, username : Annotated[StrictStr, Field(..., description="The name that needs to be deleted")], **kwargs) -> ApiResponse: # noqa: E501
async def delete_user_with_http_info(self, username : Annotated[StrictStr, Field(..., description="The name that needs to be deleted")], **kwargs) -> ApiResponse: # noqa: E501
"""Delete user # noqa: E501
This can only be done by the logged in user. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.delete_user_with_http_info(username, async_req=True)
>>> result = thread.get()
:param username: The name that needs to be deleted (required)
:type username: 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.
@ -597,7 +495,6 @@ class UserApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -639,7 +536,7 @@ class UserApi:
_response_types_map = {}
return self.api_client.call_api(
return await self.api_client.call_api(
'/user/{username}', 'DELETE',
_path_params,
_query_params,
@ -649,36 +546,20 @@ class UserApi:
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'))
@overload
async def get_user_by_name(self, username : Annotated[StrictStr, Field(..., description="The name that needs to be fetched. Use user1 for testing.")], **kwargs) -> User: # noqa: E501
...
@overload
def get_user_by_name(self, username : Annotated[StrictStr, Field(..., description="The name that needs to be fetched. Use user1 for testing.")], async_req: Optional[bool]=True, **kwargs) -> User: # noqa: E501
...
@validate_arguments
def get_user_by_name(self, username : Annotated[StrictStr, Field(..., description="The name that needs to be fetched. Use user1 for testing.")], async_req: Optional[bool]=None, **kwargs) -> Union[User, Awaitable[User]]: # noqa: E501
async def get_user_by_name(self, username : Annotated[StrictStr, Field(..., description="The name that needs to be fetched. Use user1 for testing.")], **kwargs) -> User: # noqa: E501
"""Get user by user name # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.get_user_by_name(username, async_req=True)
>>> result = thread.get()
:param username: The name that needs to be fetched. Use user1 for testing. (required)
:type username: 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
@ -692,25 +573,16 @@ class UserApi:
if '_preload_content' in kwargs:
message = "Error! Please call the get_user_by_name_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.get_user_by_name_with_http_info(username, **kwargs) # noqa: E501
return await self.get_user_by_name_with_http_info(username, **kwargs) # noqa: E501
@validate_arguments
def get_user_by_name_with_http_info(self, username : Annotated[StrictStr, Field(..., description="The name that needs to be fetched. Use user1 for testing.")], **kwargs) -> ApiResponse: # noqa: E501
async def get_user_by_name_with_http_info(self, username : Annotated[StrictStr, Field(..., description="The name that needs to be fetched. Use user1 for testing.")], **kwargs) -> ApiResponse: # noqa: E501
"""Get user by user name # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.get_user_by_name_with_http_info(username, async_req=True)
>>> result = thread.get()
:param username: The name that needs to be fetched. Use user1 for testing. (required)
:type username: 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.
@ -741,7 +613,6 @@ class UserApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -791,7 +662,7 @@ class UserApi:
'404': None,
}
return self.api_client.call_api(
return await self.api_client.call_api(
'/user/{username}', 'GET',
_path_params,
_query_params,
@ -801,38 +672,22 @@ class UserApi:
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'))
@overload
async def login_user(self, username : Annotated[StrictStr, Field(..., description="The user name for login")], password : Annotated[StrictStr, Field(..., description="The password for login in clear text")], **kwargs) -> str: # noqa: E501
...
@overload
def login_user(self, username : Annotated[StrictStr, Field(..., description="The user name for login")], password : Annotated[StrictStr, Field(..., description="The password for login in clear text")], async_req: Optional[bool]=True, **kwargs) -> str: # noqa: E501
...
@validate_arguments
def login_user(self, username : Annotated[StrictStr, Field(..., description="The user name for login")], password : Annotated[StrictStr, Field(..., description="The password for login in clear text")], async_req: Optional[bool]=None, **kwargs) -> Union[str, Awaitable[str]]: # noqa: E501
async def login_user(self, username : Annotated[StrictStr, Field(..., description="The user name for login")], password : Annotated[StrictStr, Field(..., description="The password for login in clear text")], **kwargs) -> str: # noqa: E501
"""Logs user into the system # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.login_user(username, password, async_req=True)
>>> result = thread.get()
:param username: The user name for login (required)
:type username: str
:param password: The password for login in clear text (required)
:type password: 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
@ -846,27 +701,18 @@ class UserApi:
if '_preload_content' in kwargs:
message = "Error! Please call the login_user_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.login_user_with_http_info(username, password, **kwargs) # noqa: E501
return await self.login_user_with_http_info(username, password, **kwargs) # noqa: E501
@validate_arguments
def login_user_with_http_info(self, username : Annotated[StrictStr, Field(..., description="The user name for login")], password : Annotated[StrictStr, Field(..., description="The password for login in clear text")], **kwargs) -> ApiResponse: # noqa: E501
async def login_user_with_http_info(self, username : Annotated[StrictStr, Field(..., description="The user name for login")], password : Annotated[StrictStr, Field(..., description="The password for login in clear text")], **kwargs) -> ApiResponse: # noqa: E501
"""Logs user into the system # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.login_user_with_http_info(username, password, async_req=True)
>>> result = thread.get()
:param username: The user name for login (required)
:type username: str
:param password: The password for login in clear text (required)
:type password: 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.
@ -898,7 +744,6 @@ class UserApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -950,7 +795,7 @@ class UserApi:
'400': None,
}
return self.api_client.call_api(
return await self.api_client.call_api(
'/user/login', 'GET',
_path_params,
_query_params,
@ -960,34 +805,18 @@ class UserApi:
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'))
@overload
async def logout_user(self, **kwargs) -> None: # noqa: E501
...
@overload
def logout_user(self, async_req: Optional[bool]=True, **kwargs) -> None: # noqa: E501
...
@validate_arguments
def logout_user(self, async_req: Optional[bool]=None, **kwargs) -> Union[None, Awaitable[None]]: # noqa: E501
async def logout_user(self, **kwargs) -> None: # noqa: E501
"""Logs out current logged in user session # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.logout_user(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
@ -1001,23 +830,14 @@ class UserApi:
if '_preload_content' in kwargs:
message = "Error! Please call the logout_user_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.logout_user_with_http_info(**kwargs) # noqa: E501
return await self.logout_user_with_http_info(**kwargs) # noqa: E501
@validate_arguments
def logout_user_with_http_info(self, **kwargs) -> ApiResponse: # noqa: E501
async def logout_user_with_http_info(self, **kwargs) -> ApiResponse: # noqa: E501
"""Logs out current logged in user session # noqa: E501
# noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.logout_user_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.
@ -1047,7 +867,6 @@ class UserApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -1086,7 +905,7 @@ class UserApi:
_response_types_map = {}
return self.api_client.call_api(
return await self.api_client.call_api(
'/user/logout', 'GET',
_path_params,
_query_params,
@ -1096,38 +915,22 @@ class UserApi:
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'))
@overload
async def update_user(self, username : Annotated[StrictStr, Field(..., description="name that need to be deleted")], user : Annotated[User, Field(..., description="Updated user object")], **kwargs) -> None: # noqa: E501
...
@overload
def update_user(self, username : Annotated[StrictStr, Field(..., description="name that need to be deleted")], user : Annotated[User, Field(..., description="Updated user object")], async_req: Optional[bool]=True, **kwargs) -> None: # noqa: E501
...
@validate_arguments
def update_user(self, username : Annotated[StrictStr, Field(..., description="name that need to be deleted")], user : Annotated[User, Field(..., description="Updated user object")], async_req: Optional[bool]=None, **kwargs) -> Union[None, Awaitable[None]]: # noqa: E501
async def update_user(self, username : Annotated[StrictStr, Field(..., description="name that need to be deleted")], user : Annotated[User, Field(..., description="Updated user object")], **kwargs) -> None: # noqa: E501
"""Updated user # noqa: E501
This can only be done by the logged in user. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.update_user(username, user, async_req=True)
>>> result = thread.get()
:param username: name that need to be deleted (required)
:type username: str
:param user: Updated user object (required)
:type user: User
: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
@ -1141,27 +944,18 @@ class UserApi:
if '_preload_content' in kwargs:
message = "Error! Please call the update_user_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.update_user_with_http_info(username, user, **kwargs) # noqa: E501
return await self.update_user_with_http_info(username, user, **kwargs) # noqa: E501
@validate_arguments
def update_user_with_http_info(self, username : Annotated[StrictStr, Field(..., description="name that need to be deleted")], user : Annotated[User, Field(..., description="Updated user object")], **kwargs) -> ApiResponse: # noqa: E501
async def update_user_with_http_info(self, username : Annotated[StrictStr, Field(..., description="name that need to be deleted")], user : Annotated[User, Field(..., description="Updated user object")], **kwargs) -> ApiResponse: # noqa: E501
"""Updated user # noqa: E501
This can only be done by the logged in user. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.update_user_with_http_info(username, user, async_req=True)
>>> result = thread.get()
:param username: name that need to be deleted (required)
:type username: str
:param user: Updated user object (required)
:type user: User
: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.
@ -1193,7 +987,6 @@ class UserApi:
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
@ -1245,7 +1038,7 @@ class UserApi:
_response_types_map = {}
return self.api_client.call_api(
return await self.api_client.call_api(
'/user/{username}', 'PUT',
_path_params,
_query_params,
@ -1255,7 +1048,6 @@ class UserApi:
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'),

93
samples/openapi3/client/petstore/python-aiohttp/petstore_api/api_client.py
View File

@ -17,7 +17,6 @@ import datetime
from dateutil.parser import parse
import json
import mimetypes
from multiprocessing.pool import ThreadPool
import os
import re
import tempfile
@ -45,8 +44,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,12 +60,11 @@ class ApiClient:
_pool = None
def __init__(self, configuration=None, header_name=None, header_value=None,
cookie=None, pool_threads=1) -> 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 = {}
@ -87,22 +83,6 @@ class ApiClient:
async def close(self):
await self.rest_client.close()
if self._pool:
self._pool.close()
self._pool.join()
self._pool = None
if hasattr(atexit, 'unregister'):
atexit.unregister(self.close)
@property
def pool(self):
"""Create thread pool on first request
avoids instantiating unused threadpool for blocking clients.
"""
if self._pool is None:
atexit.register(self.close)
self._pool = ThreadPool(self.pool_threads)
return self._pool
@property
def user_agent(self):
@ -223,7 +203,7 @@ class ApiClient:
self.last_response = response_data
return_data = None # assuming derialization is not needed
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)
@ -358,17 +338,15 @@ class ApiClient:
else:
return self.__deserialize_model(data, klass)
def call_api(self, resource_path, method,
async def call_api(self, resource_path, method,
path_params=None, query_params=None, header_params=None,
body=None, post_params=None, files=None,
response_types_map=None, auth_settings=None,
async_req=None, _return_http_data_only=None,
_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.
@ -382,7 +360,6 @@ class ApiClient:
: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
@ -400,58 +377,52 @@ class ApiClient:
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.
The response.
"""
if not async_req:
return self.__call_api(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)
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,
)
return await self.__call_api(*args)
return self.pool.apply_async(self.__call_api, (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))
def request(self, method, url, query_params=None, headers=None,
async 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,
return await 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,
return await 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,
return await 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,
return await self.rest_client.post_request(url,
query_params=query_params,
headers=headers,
post_params=post_params,
@ -459,7 +430,7 @@ class ApiClient:
_request_timeout=_request_timeout,
body=body)
elif method == "PUT":
return self.rest_client.put_request(url,
return await self.rest_client.put_request(url,
query_params=query_params,
headers=headers,
post_params=post_params,
@ -467,7 +438,7 @@ class ApiClient:
_request_timeout=_request_timeout,
body=body)
elif method == "PATCH":
return self.rest_client.patch_request(url,
return await self.rest_client.patch_request(url,
query_params=query_params,
headers=headers,
post_params=post_params,
@ -475,7 +446,7 @@ class ApiClient:
_request_timeout=_request_timeout,
body=body)
elif method == "DELETE":
return self.rest_client.delete_request(url,
return await self.rest_client.delete_request(url,
query_params=query_params,
headers=headers,
_preload_content=_preload_content,

5
samples/openapi3/client/petstore/python-aiohttp/tests/test_api_client.py
View File

@ -15,13 +15,8 @@ class TestApiClient(unittest.TestCase):
async def test_context_manager_closes_client(self):
async with petstore_api.ApiClient() as client:
# thread pool
self.assertIsNotNone(client.pool)
pool_ref = weakref.ref(client._pool)
self.assertIsNotNone(pool_ref())
# pool_manager
self.assertFalse(client.rest_client.pool_manager.closed)
rest_pool_ref = client.rest_client.pool_manager
self.assertIsNone(pool_ref())
self.assertTrue(rest_pool_ref.closed)

3
samples/openapi3/client/petstore/python-aiohttp/tests/util.py
View File

@ -11,7 +11,8 @@ def id_gen(bits=32):
def async_test(f):
def wrapper(*args, **kwargs):
coro = asyncio.coroutine(f)
# coro = asyncio.coroutine(f)
coro = f
future = coro(*args, **kwargs)
loop = asyncio.get_event_loop()
loop.run_until_complete(future)

41
samples/openapi3/client/petstore/python/petstore_api/api_client.py
View File

@ -222,7 +222,7 @@ class ApiClient:
self.last_response = response_data
return_data = None # assuming derialization is not needed
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)
@ -405,27 +405,28 @@ class ApiClient:
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(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)
return self.__call_api(*args)
return self.pool.apply_async(self.__call_api, (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))
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,