Merge pull request #984 from geekerzp/python-enum

[Python] Support enum for model property
This commit is contained in:
wing328 2015-07-22 16:48:34 +08:00
commit cde8d7ba4c
24 changed files with 2257 additions and 1405 deletions

View File

@ -117,7 +117,7 @@ public class PythonClientCodegen extends DefaultCodegen implements CodegenConfig
@Override
public String escapeReservedWord(String name) {
return "_" + name;
return name + "_";
}
@Override
@ -160,8 +160,7 @@ public class PythonClientCodegen extends DefaultCodegen implements CodegenConfig
}
public String toDefaultValue(Property p) {
// TODO: Support Python def value
return "null";
return "None";
}
@Override
@ -183,6 +182,9 @@ public class PythonClientCodegen extends DefaultCodegen implements CodegenConfig
name = escapeReservedWord(name);
}
// remove leading underscore
name = name.replaceAll("^_*", "");
return name;
}

View File

@ -1,5 +1,8 @@
from __future__ import absolute_import
# import apis into api package
{{#apiInfo}}{{#apis}}from .{{classVarName}} import {{classname}}
{{/apis}}{{/apiInfo}}
{{#apiInfo}}
{{#apis}}
from .{{classVarName}} import {{classname}}
{{/apis}}
{{/apiInfo}}

View File

@ -1,5 +1,5 @@
from __future__ import absolute_import
# import models into model package
{{#models}}{{#model}}from .{{classVarName}} import {{classname}}
{{/model}}{{/models}}
{{#models}}{{#model}}from .{{classVarName}} import {{classname}}{{/model}}
{{/models}}

View File

@ -17,7 +17,8 @@ Copyright 2015 SmartBear Software
See the License for the specific language governing permissions and
limitations under the License.
NOTE: This class is auto generated by the swagger code generator program. Do not edit the class manually.
NOTE: This class is auto generated by the swagger code generator program.
Do not edit the class manually.
"""
from __future__ import absolute_import
@ -30,6 +31,7 @@ from six import iteritems
from ..configuration import Configuration
from ..api_client import ApiClient
{{#operations}}
class {{classname}}(object):
@ -41,38 +43,48 @@ class {{classname}}(object):
if not config.api_client:
config.api_client = ApiClient('{{basePath}}')
self.api_client = config.api_client
{{#operation}}
{{#operation}}
def {{nickname}}(self, {{#allParams}}{{#required}}{{paramName}}, {{/required}}{{/allParams}}**kwargs):
"""
{{{summary}}}
{{{notes}}}
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.{{nickname}}({{#allParams}}{{#required}}{{paramName}}, {{/required}}{{/allParams}}callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
{{#allParams}}:param {{dataType}} {{paramName}}: {{{description}}} {{#required}}(required){{/required}}{{#optional}}(optional){{/optional}}
{{/allParams}}
:param callback function: The callback function
for asynchronous request. (optional)
{{#allParams}}
:param {{dataType}} {{paramName}}: {{{description}}}{{#required}} (required){{/required}}{{#optional}}(optional){{/optional}}
{{/allParams}}
:return: {{#returnType}}{{returnType}}{{/returnType}}{{^returnType}}None{{/returnType}}
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
{{#allParams}}{{#required}}
{{#allParams}}
{{#required}}
# verify the required parameter '{{paramName}}' is set
if {{paramName}} is None:
raise ValueError("Missing the required parameter `{{paramName}}` when calling `{{nickname}}`")
{{/required}}{{/allParams}}
{{/required}}
{{/allParams}}
all_params = [{{#allParams}}'{{paramName}}'{{#hasMore}}, {{/hasMore}}{{/allParams}}]
all_params.append('callback')
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method {{nickname}}" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method {{nickname}}" % key
)
params[key] = val
del params['kwargs']
@ -80,45 +92,59 @@ class {{classname}}(object):
method = '{{httpMethod}}'
path_params = {}
{{#pathParams}}
{{#pathParams}}
if '{{paramName}}' in params:
path_params['{{baseName}}'] = params['{{paramName}}']
{{/pathParams}}
{{/pathParams}}
query_params = {}
{{#queryParams}}
{{#queryParams}}
if '{{paramName}}' in params:
query_params['{{baseName}}'] = params['{{paramName}}']
{{/queryParams}}
{{/queryParams}}
header_params = {}
{{#headerParams}}
{{#headerParams}}
if '{{paramName}}' in params:
header_params['{{baseName}}'] = params['{{paramName}}']
{{/headerParams}}
{{/headerParams}}
form_params = {}
files = {}
{{#formParams}}
{{#formParams}}
if '{{paramName}}' in params:
{{#notFile}}form_params['{{baseName}}'] = params['{{paramName}}']{{/notFile}}{{#isFile}}files['{{baseName}}'] = params['{{paramName}}']{{/isFile}}
{{/formParams}}
{{/formParams}}
body_params = None
{{#bodyParam}}
{{#bodyParam}}
if '{{paramName}}' in params:
body_params = params['{{paramName}}']
{{/bodyParam}}
{{/bodyParam}}
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept([{{#produces}}'{{mediaType}}'{{#hasMore}}, {{/hasMore}}{{/produces}}])
header_params['Accept'] = self.api_client.\
select_header_accept([{{#produces}}'{{mediaType}}'{{#hasMore}}, {{/hasMore}}{{/produces}}])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([{{#consumes}}'{{mediaType}}'{{#hasMore}}, {{/hasMore}}{{/consumes}}])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([{{#consumes}}'{{mediaType}}'{{#hasMore}}, {{/hasMore}}{{/consumes}}])
# Authentication setting
auth_settings = [{{#authMethods}}'{{name}}'{{#hasMore}}, {{/hasMore}}{{/authMethods}}]
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type={{#returnType}}'{{returnType}}'{{/returnType}}{{^returnType}}None{{/returnType}}, auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type={{#returnType}}'{{returnType}}'{{/returnType}}{{^returnType}}None{{/returnType}},
auth_settings=auth_settings,
callback=params.get('callback'))
return response
{{/operation}}
{{/operation}}
{{/operations}}

View File

@ -27,415 +27,480 @@ from datetime import date
from six import iteritems
try:
# for python3
from urllib.parse import quote
# for python3
from urllib.parse import quote
except ImportError:
# for python2
from urllib import quote
# for python2
from urllib import quote
from .configuration import Configuration
class ApiClient(object):
"""
Generic API client for Swagger client library builds
:param host: The base path for the server to call
:param header_name: a header to pass when making calls to the API
:param header_value: a header value to pass when making calls to the API
"""
def __init__(self, host=Configuration().host, header_name=None, header_value=None):
self.default_headers = {}
if header_name is not None:
self.default_headers[header_name] = header_value
self.host = host
self.cookie = None
# Set default User-Agent.
self.user_agent = 'Python-Swagger'
@property
def user_agent(self):
return self.default_headers['User-Agent']
@user_agent.setter
def user_agent(self, value):
self.default_headers['User-Agent'] = value
def set_default_header(self, header_name, header_value):
self.default_headers[header_name] = header_value
def __call_api(self, resource_path, method, path_params=None, query_params=None, header_params=None,
body=None, post_params=None, files=None, response_type=None, auth_settings=None, callback=None):
# headers parameters
header_params = header_params or {}
header_params.update(self.default_headers)
if self.cookie:
header_params['Cookie'] = self.cookie
if header_params:
header_params = self.sanitize_for_serialization(header_params)
# path parameters
if path_params:
path_params = self.sanitize_for_serialization(path_params)
for k, v in iteritems(path_params):
replacement = quote(str(self.to_path_value(v)))
resource_path = resource_path.replace('{' + k + '}', replacement)
# query parameters
if query_params:
query_params = self.sanitize_for_serialization(query_params)
query_params = {k: self.to_path_value(v) for k, v in iteritems(query_params)}
# post parameters
if post_params:
post_params = self.prepare_post_parameters(post_params, files)
post_params = self.sanitize_for_serialization(post_params)
# auth setting
self.update_params_for_auth(header_params, query_params, auth_settings)
# body
if body:
body = self.sanitize_for_serialization(body)
# request url
url = self.host + resource_path
# perform request and return response
response_data = self.request(method, url, query_params=query_params, headers=header_params,
post_params=post_params, body=body)
self.last_response = response_data
# deserialize response data
if response_type:
deserialized_data = self.deserialize(response_data, response_type)
else:
deserialized_data = None
if callback:
callback(deserialized_data)
else:
return deserialized_data
def to_path_value(self, obj):
"""
Convert a string or object to a path-friendly value
Generic API client for Swagger client library builds
:param obj: object or string value
:return string: quoted value
:param host: The base path for the server to call
:param header_name: a header to pass when making calls to the API
:param header_value: a header value to pass when making calls to the API
"""
if type(obj) == list:
return ','.join(obj)
else:
return str(obj)
def __init__(self, host=Configuration().host,
header_name=None, header_value=None):
def sanitize_for_serialization(self, obj):
"""
Sanitize an object for Request.
self.default_headers = {}
if header_name is not None:
self.default_headers[header_name] = header_value
self.host = host
self.cookie = None
# Set default User-Agent.
self.user_agent = 'Python-Swagger'
If obj is None, return None.
If obj is str, int, float, bool, return directly.
If obj is datetime.datetime, datetime.date convert to string in iso8601 format.
If obj is list, santize each element in the list.
If obj is dict, return the dict.
If obj is swagger model, return the properties dict.
"""
if isinstance(obj, type(None)):
return None
elif isinstance(obj, (str, int, float, bool, tuple)):
return obj
elif isinstance(obj, list):
return [self.sanitize_for_serialization(sub_obj) for sub_obj in obj]
elif isinstance(obj, (datetime, date)):
return obj.isoformat()
else:
if isinstance(obj, dict):
obj_dict = obj
else:
# Convert model obj to dict except attributes `swagger_types`, `attribute_map`
# and attributes which value is not None.
# Convert attribute name to json key in model definition for request.
obj_dict = {obj.attribute_map[key]: val
for key, val in iteritems(obj.__dict__)
if key != 'swagger_types' and key != 'attribute_map' and val is not None}
return {key: self.sanitize_for_serialization(val)
for key, val in iteritems(obj_dict)}
@property
def user_agent(self):
return self.default_headers['User-Agent']
def deserialize(self, response, response_type):
"""
Derialize response into an object.
@user_agent.setter
def user_agent(self, value):
self.default_headers['User-Agent'] = value
:param response: RESTResponse object to be deserialized
:param response_type: class literal for deserialzied object, or string of class name
def set_default_header(self, header_name, header_value):
self.default_headers[header_name] = header_value
:return: deserialized object
"""
# handle file downloading - save response body into a tmp file and return the instance
if "file" == response_type:
return self.__deserialize_file(response)
def __call_api(self, resource_path, method,
path_params=None, query_params=None, header_params=None,
body=None, post_params=None, files=None,
response_type=None, auth_settings=None, callback=None):
# fetch data from response object
try:
data = json.loads(response.data)
except ValueError:
data = response.data
# headers parameters
header_params = header_params or {}
header_params.update(self.default_headers)
if self.cookie:
header_params['Cookie'] = self.cookie
if header_params:
header_params = self.sanitize_for_serialization(header_params)
return self.__deserialize(data, response_type)
# path parameters
if path_params:
path_params = self.sanitize_for_serialization(path_params)
for k, v in iteritems(path_params):
replacement = quote(str(self.to_path_value(v)))
resource_path = resource_path.\
replace('{' + k + '}', replacement)
def __deserialize(self, data, klass):
"""
:param data: dict, list or str
:param klass: class literal, or string of class name
# query parameters
if query_params:
query_params = self.sanitize_for_serialization(query_params)
query_params = {k: self.to_path_value(v)
for k, v in iteritems(query_params)}
:return: object
"""
if data is None:
return None
# post parameters
if post_params:
post_params = self.prepare_post_parameters(post_params, files)
post_params = self.sanitize_for_serialization(post_params)
if type(klass) == str:
if 'list[' in klass:
sub_kls = re.match('list\[(.*)\]', klass).group(1)
return [self.__deserialize(sub_data, sub_kls) for sub_data in data]
# auth setting
self.update_params_for_auth(header_params, query_params, auth_settings)
if 'dict(' in klass:
sub_kls = re.match('dict\((.*), (.*)\)', klass).group(2)
return {k: self.__deserialize(v, sub_kls) for k, v in iteritems(data)}
# body
if body:
body = self.sanitize_for_serialization(body)
# convert str to class
# for native types
if klass in ['int', 'float', 'str', 'bool', "date", 'datetime', "object"]:
klass = eval(klass)
# for model types
else:
klass = eval('models.' + klass)
# request url
url = self.host + resource_path
if klass in [int, float, str, bool]:
return self.__deserialize_primitive(data, klass)
elif klass == object:
return self.__deserialize_object()
elif klass == date:
return self.__deserialize_date(data)
elif klass == datetime:
return self.__deserialize_datatime(data)
else:
return self.__deserialize_model(data, klass)
# perform request and return response
response_data = self.request(method, url,
query_params=query_params,
headers=header_params,
post_params=post_params, body=body)
def call_api(self, resource_path, method,
path_params=None, query_params=None, header_params=None,
body=None, post_params=None, files=None,
response_type=None, auth_settings=None, callback=None):
"""
Perform http request and return deserialized data
self.last_response = response_data
:param resource_path: Path to method endpoint.
:param method: Method to call.
:param path_params: Path parameters in the url.
:param query_params: Query parameters in the url.
:param header_params: Header parameters to be placed in the request header.
:param body: Request body.
:param post_params dict: Request post form parameters, for `application/x-www-form-urlencoded`, `multipart/form-data`.
:param auth_settings list: Auth Settings names for the request.
:param response: Response data type.
:param files dict: key -> filename, value -> filepath, for `multipart/form-data`.
:param callback function: Callback function for asynchronous request.
If provide this parameter, the request will be called asynchronously.
:return:
If provide parameter callback, the request will be called asynchronously.
The method will return the request thread.
If parameter callback is None, then the method will return the response directly.
"""
if callback is None:
return self.__call_api(resource_path, method,
path_params, query_params, header_params,
body, post_params, files,
response_type, auth_settings, callback)
else:
thread = threading.Thread(target=self.__call_api,
args=(resource_path, method,
path_params, query_params, header_params,
body, post_params, files,
response_type, auth_settings, callback))
thread.start()
return thread
def request(self, method, url, query_params=None, headers=None,
post_params=None, body=None):
"""
Perform http request using RESTClient.
"""
if method == "GET":
return RESTClient.GET(url, query_params=query_params, headers=headers)
elif method == "HEAD":
return RESTClient.HEAD(url, query_params=query_params, headers=headers)
elif method == "POST":
return RESTClient.POST(url, headers=headers, post_params=post_params, body=body)
elif method == "PUT":
return RESTClient.PUT(url, headers=headers, post_params=post_params, body=body)
elif method == "PATCH":
return RESTClient.PATCH(url, headers=headers, post_params=post_params, body=body)
elif method == "DELETE":
return RESTClient.DELETE(url, query_params=query_params, headers=headers)
else:
raise ValueError("http method must be `GET`, `HEAD`, `POST`, `PATCH`, `PUT` or `DELETE`")
def prepare_post_parameters(self, post_params=None, files=None):
params = {}
if post_params:
params.update(post_params)
if files:
for k, v in iteritems(files):
if v:
with open(v, 'rb') as f:
filename = os.path.basename(f.name)
filedata = f.read()
mimetype = mimetypes.guess_type(filename)[0] or 'application/octet-stream'
params[k] = tuple([filename, filedata, mimetype])
return params
def select_header_accept(self, accepts):
"""
Return `Accept` based on an array of accepts provided
"""
if not accepts:
return
accepts = list(map(lambda x: x.lower(), accepts))
if 'application/json' in accepts:
return 'application/json'
else:
return ', '.join(accepts)
def select_header_content_type(self, content_types):
"""
Return `Content-Type` baseed on an array of content_types provided
"""
if not content_types:
return 'application/json'
content_types = list(map(lambda x: x.lower(), content_types))
if 'application/json' in content_types:
return 'application/json'
else:
return content_types[0]
def update_params_for_auth(self, headers, querys, auth_settings):
"""
Update header and query params based on authentication setting
"""
config = Configuration()
if not auth_settings:
return
for auth in auth_settings:
auth_setting = config.auth_settings().get(auth)
if auth_setting:
if auth_setting['in'] == 'header':
headers[auth_setting['key']] = auth_setting['value']
elif auth_setting['in'] == 'query':
querys[auth_setting['key']] = auth_setting['value']
# deserialize response data
if response_type:
deserialized_data = self.deserialize(response_data, response_type)
else:
raise ValueError('Authentication token must be in `query` or `header`')
deserialized_data = None
def __deserialize_file(self, response):
"""
Save response body into a file in (the defined) temporary folder, using the filename
from the `Content-Disposition` header if provided, otherwise a random filename.
if callback:
callback(deserialized_data)
else:
return deserialized_data
:param response: RESTResponse
:return: file path
"""
fd, path = tempfile.mkstemp(dir=configuration.temp_folder_path)
os.close(fd)
os.remove(path)
def to_path_value(self, obj):
"""
Convert a string or object to a path-friendly value
content_disposition = response.getheader("Content-Disposition")
if content_disposition:
filename = re.search(r'filename=[\'"]?([^\'"\s]+)[\'"]?', content_disposition).group(1)
path = os.path.join(os.path.dirname(path), filename)
:param obj: object or string value
with open(path, "w") as f:
f.write(response.data)
:return string: quoted value
"""
if type(obj) == list:
return ','.join(obj)
else:
return str(obj)
return path
def sanitize_for_serialization(self, obj):
"""
Sanitize an object for Request.
def __deserialize_primitive(self, data, klass):
"""
Deserialize string to primitive type
If obj is None, return None.
If obj is str, int, float, bool, return directly.
If obj is datetime.datetime, datetime.date
convert to string in iso8601 format.
If obj is list, santize each element in the list.
If obj is dict, return the dict.
If obj is swagger model, return the properties dict.
"""
if isinstance(obj, type(None)):
return None
elif isinstance(obj, (str, int, float, bool, tuple)):
return obj
elif isinstance(obj, list):
return [self.sanitize_for_serialization(sub_obj)
for sub_obj in obj]
elif isinstance(obj, (datetime, date)):
return obj.isoformat()
else:
if isinstance(obj, dict):
obj_dict = obj
else:
# Convert model obj to dict except
# attributes `swagger_types`, `attribute_map`
# and attributes which value is not None.
# Convert attribute name to json key in
# model definition for request.
obj_dict = {obj.attribute_map[key[1:]]: val
for key, val in iteritems(obj.__dict__)
if key != 'swagger_types'
and key != 'attribute_map'
and val is not None}
:param data: str
:param klass: class literal
return {key: self.sanitize_for_serialization(val)
for key, val in iteritems(obj_dict)}
:return: int, float, str, bool
"""
try:
value = klass(data)
except UnicodeEncodeError:
value = unicode(data)
except TypeError:
value = data
return value
def deserialize(self, response, response_type):
"""
Derialize response into an object.
def __deserialize_object(self):
"""
Deserialize empty object
"""
return object()
:param response: RESTResponse object to be deserialized
:param response_type: class literal for
deserialzied object, or string of class name
def __deserialize_date(self, string):
"""
Deserialize string to date
:return: deserialized object
"""
# handle file downloading
# save response body into a tmp file and return the instance
if "file" == response_type:
return self.__deserialize_file(response)
:param string: str
:return: date
"""
try:
from dateutil.parser import parse
return parse(string).date()
except ImportError:
return string
except ValueError:
raise ApiException(status=0, reason="Failed to parse `{0}` into a date object".format(string))
# fetch data from response object
try:
data = json.loads(response.data)
except ValueError:
data = response.data
def __deserialize_datatime(self, string):
"""
Deserialize string to datetime.
return self.__deserialize(data, response_type)
The string should be in iso8601 datetime format.
def __deserialize(self, data, klass):
"""
:param data: dict, list or str
:param klass: class literal, or string of class name
:param string: str
:return: datetime
"""
try:
from dateutil.parser import parse
return parse(string)
except ImportError:
return string
except ValueError:
raise ApiException(status=0, reason="Failed to parse `{0}` into a datetime object".format(string))
:return: object
"""
if data is None:
return None
def __deserialize_model(self, data, klass):
"""
Deserialize list or dict to model
if type(klass) == str:
if 'list[' in klass:
sub_kls = re.match('list\[(.*)\]', klass).group(1)
return [self.__deserialize(sub_data, sub_kls)
for sub_data in data]
:param data: dict, list
:param klass: class literal
"""
instance = klass()
if 'dict(' in klass:
sub_kls = re.match('dict\((.*), (.*)\)', klass).group(2)
return {k: self.__deserialize(v, sub_kls)
for k, v in iteritems(data)}
for attr, attr_type in iteritems(instance.swagger_types):
if data is not None \
and instance.attribute_map[attr] in data\
and isinstance(data, (list, dict)):
value = data[instance.attribute_map[attr]]
setattr(instance, attr, self.__deserialize(value, attr_type))
# convert str to class
# for native types
if klass in ['int', 'float', 'str', 'bool',
"date", 'datetime', "object"]:
klass = eval(klass)
# for model types
else:
klass = eval('models.' + klass)
return instance
if klass in [int, float, str, bool]:
return self.__deserialize_primitive(data, klass)
elif klass == object:
return self.__deserialize_object()
elif klass == date:
return self.__deserialize_date(data)
elif klass == datetime:
return self.__deserialize_datatime(data)
else:
return self.__deserialize_model(data, klass)
def call_api(self, resource_path, method,
path_params=None, query_params=None, header_params=None,
body=None, post_params=None, files=None,
response_type=None, auth_settings=None, callback=None):
"""
Perform http request and return deserialized data
:param resource_path: Path to method endpoint.
:param method: Method to call.
:param path_params: Path parameters in the url.
:param query_params: Query parameters in the url.
:param header_params: Header parameters to be
placed in the request header.
:param body: Request body.
:param post_params dict: Request post form parameters,
for `application/x-www-form-urlencoded`, `multipart/form-data`.
:param auth_settings list: Auth Settings names for the request.
:param response: Response data type.
:param files dict: key -> filename, value -> filepath,
for `multipart/form-data`.
:param callback function: Callback function for asynchronous request.
If provide this parameter,
the request will be called asynchronously.
:return:
If provide parameter callback,
the request will be called asynchronously.
The method will return the request thread.
If parameter callback is None,
then the method will return the response directly.
"""
if callback is None:
return self.__call_api(resource_path, method,
path_params, query_params, header_params,
body, post_params, files,
response_type, auth_settings, callback)
else:
thread = threading.Thread(target=self.__call_api,
args=(resource_path, method,
path_params, query_params,
header_params, body,
post_params, files,
response_type, auth_settings,
callback))
thread.start()
return thread
def request(self, method, url, query_params=None, headers=None,
post_params=None, body=None):
"""
Perform http request using RESTClient.
"""
if method == "GET":
return RESTClient.GET(url,
query_params=query_params,
headers=headers)
elif method == "HEAD":
return RESTClient.HEAD(url,
query_params=query_params,
headers=headers)
elif method == "POST":
return RESTClient.POST(url,
headers=headers,
post_params=post_params,
body=body)
elif method == "PUT":
return RESTClient.PUT(url,
headers=headers,
post_params=post_params,
body=body)
elif method == "PATCH":
return RESTClient.PATCH(url,
headers=headers,
post_params=post_params,
body=body)
elif method == "DELETE":
return RESTClient.DELETE(url,
query_params=query_params,
headers=headers)
else:
raise ValueError(
"http method must be `GET`, `HEAD`,"
" `POST`, `PATCH`, `PUT` or `DELETE`."
)
def prepare_post_parameters(self, post_params=None, files=None):
params = {}
if post_params:
params.update(post_params)
if files:
for k, v in iteritems(files):
if not v:
continue
with open(v, 'rb') as f:
filename = os.path.basename(f.name)
filedata = f.read()
mimetype = mimetypes.\
guess_type(filename)[0] or 'application/octet-stream'
params[k] = tuple([filename, filedata, mimetype])
return params
def select_header_accept(self, accepts):
"""
Return `Accept` based on an array of accepts provided
"""
if not accepts:
return
accepts = list(map(lambda x: x.lower(), accepts))
if 'application/json' in accepts:
return 'application/json'
else:
return ', '.join(accepts)
def select_header_content_type(self, content_types):
"""
Return `Content-Type` baseed on an array of content_types provided
"""
if not content_types:
return 'application/json'
content_types = list(map(lambda x: x.lower(), content_types))
if 'application/json' in content_types:
return 'application/json'
else:
return content_types[0]
def update_params_for_auth(self, headers, querys, auth_settings):
"""
Update header and query params based on authentication setting
"""
config = Configuration()
if not auth_settings:
return
for auth in auth_settings:
auth_setting = config.auth_settings().get(auth)
if auth_setting:
if auth_setting['in'] == 'header':
headers[auth_setting['key']] = auth_setting['value']
elif auth_setting['in'] == 'query':
querys[auth_setting['key']] = auth_setting['value']
else:
raise ValueError(
'Authentication token must be in `query` or `header`'
)
def __deserialize_file(self, response):
"""
Save response body into a file in (the defined) temporary folder,
using the filename from the `Content-Disposition` header if provided,
otherwise a random filename.
:param response: RESTResponse
:return: file path
"""
config = Configuration()
fd, path = tempfile.mkstemp(dir=config.temp_folder_path)
os.close(fd)
os.remove(path)
content_disposition = response.getheader("Content-Disposition")
if content_disposition:
filename = re.\
search(r'filename=[\'"]?([^\'"\s]+)[\'"]?', content_disposition).\
group(1)
path = os.path.join(os.path.dirname(path), filename)
with open(path, "w") as f:
f.write(response.data)
return path
def __deserialize_primitive(self, data, klass):
"""
Deserialize string to primitive type
:param data: str
:param klass: class literal
:return: int, float, str, bool
"""
try:
value = klass(data)
except UnicodeEncodeError:
value = unicode(data)
except TypeError:
value = data
return value
def __deserialize_object(self):
"""
Deserialize empty object
"""
return object()
def __deserialize_date(self, string):
"""
Deserialize string to date
:param string: str
:return: date
"""
try:
from dateutil.parser import parse
return parse(string).date()
except ImportError:
return string
except ValueError:
raise ApiException(
status=0,
reason="Failed to parse `{0}` into a date object"
.format(string)
)
def __deserialize_datatime(self, string):
"""
Deserialize string to datetime.
The string should be in iso8601 datetime format.
:param string: str
:return: datetime
"""
try:
from dateutil.parser import parse
return parse(string)
except ImportError:
return string
except ValueError:
raise ApiException(
status=0,
reason="Failed to parse `{0}` into a datetime object".
format(string)
)
def __deserialize_model(self, data, klass):
"""
Deserialize list or dict to model
:param data: dict, list
:param klass: class literal
"""
instance = klass()
for attr, attr_type in iteritems(instance.swagger_types):
if data is not None \
and instance.attribute_map[attr] in data\
and isinstance(data, (list, dict)):
value = data[instance.attribute_map[attr]]
setattr(instance, attr, self.__deserialize(value, attr_type))
return instance

View File

@ -5,6 +5,7 @@ import httplib
import sys
import logging
def singleton(cls, *args, **kw):
instances = {}
@ -23,6 +24,8 @@ class Configuration(object):
self.host = "{{basePath}}"
# Default api client
self.api_client = None
# Temp file folder
self.temp_folder_path = None
# Authentication Settings
self.api_key = {}
self.api_key_prefix = {}
@ -93,29 +96,33 @@ class Configuration(object):
def auth_settings(self):
""" Return Auth Settings for api client """
return { {{#authMethods}}{{#isApiKey}}
'{{name}}': {
'type': 'api_key',
'in': {{#isKeyInHeader}}'header'{{/isKeyInHeader}}{{#isKeyInQuery}}'query'{{/isKeyInQuery}},
'key': '{{keyParamName}}',
'value': self.get_api_key_with_prefix('{{keyParamName}}')
},
{{/isApiKey}}{{#isBasic}}
'{{name}}': {
'type': 'basic',
'in': 'header',
'key': 'Authorization',
'value': self.get_basic_auth_token()
},
{{/isBasic}}{{/authMethods}}
}
return {
{{#authMethods}}
{{#isApiKey}}
'{{name}}':
{
'type': 'api_key',
'in': {{#isKeyInHeader}}'header'{{/isKeyInHeader}}{{#isKeyInQuery}}'query'{{/isKeyInQuery}},
'key': '{{keyParamName}}',
'value': self.get_api_key_with_prefix('{{keyParamName}}')
},
{{/isApiKey}}
{{#isBasic}}
'{{name}}':
{
'type': 'basic',
'in': 'header',
'key': 'Authorization',
'value': self.get_basic_auth_token()
},
{{/isBasic}}
{{/authMethods}}
}
def to_debug_report(self):
return "Python SDK Debug Report:\n"\
"OS: {env}\n"\
"Python Version: {pyversion}\n"\
"Version of the API: {{version}}\n"\
"SDK Package Version: {{packageVersion}}".format(env=sys.platform, pyversion=sys.version)
"SDK Package Version: {{packageVersion}}".\
format(env=sys.platform, pyversion=sys.version)

View File

@ -18,19 +18,23 @@ Copyright 2015 SmartBear Software
"""
{{#models}}
{{#model}}
from pprint import pformat
from six import iteritems
class {{classname}}(object):
"""
NOTE: This class is auto generated by the swagger code generator program.
Do not edit the class manually.
"""
def __init__(self):
"""
Swagger model
:param dict swaggerTypes: The key is attribute name and the value is attribute type.
:param dict attributeMap: The key is attribute name and the value is json key in definition.
:param dict swaggerTypes: The key is attribute name
and the value is attribute type.
:param dict attributeMap: The key is attribute name
and the value is json key in definition.
"""
self.swagger_types = {
{{#vars}}'{{name}}': '{{{datatype}}}'{{#hasMore}},
@ -41,19 +45,58 @@ class {{classname}}(object):
{{#vars}}'{{name}}': '{{baseName}}'{{#hasMore}},
{{/hasMore}}{{/vars}}
}
{{#vars}}
{{#description}}# {{description}}{{/description}}
self.{{name}} = None # {{{datatype}}}
{{/vars}}
{{#vars}}
self._{{name}} = None{{#description}} # {{description}}{{/description}}
{{/vars}}
{{#vars}}
@property
def {{name}}(self):
return self._{{name}}
@{{name}}.setter
def {{name}}(self, {{name}}):
{{#isEnum}}allowed_values = [{{#allowableValues}}{{#values}}"{{{this}}}"{{^-last}}, {{/-last}}{{/values}}{{/allowableValues}}]
if {{name}} not in allowed_values:
raise ValueError(
"Invalid value for `{{name}}`, must be one of {0}"
.format(allowed_values)
)
{{/isEnum}}self._{{name}} = {{name}}
{{/vars}}
def to_dict(self):
"""
Return model properties dict
"""
result = {}
for name, prop in iteritems(self.__dict__):
if name == "attribute_map" or name == "swagger_types":
continue
if isinstance(prop, list):
result[name[1:]] = list(map(
lambda x: x.to_dict() if hasattr(x, "to_dict") else x,
prop
))
elif hasattr(prop, "to_dict"):
result[name[1:]] = prop.to_dict()
else:
result[name[1:]] = prop
return result
def to_str(self):
"""
Return model properties str
"""
return pformat(self.to_dict())
def __repr__(self):
properties = []
for p in self.__dict__:
if p != 'swaggerTypes' and p != 'attributeMap':
properties.append('{prop}={val!r}'.format(prop=p, val=self.__dict__[p]))
return '<{name} {props}>'.format(name=__name__, props=' '.join(properties))
"""
For `print` and `pprint`
"""
return self.to_str()
{{/model}}
{{/models}}

View File

@ -51,6 +51,7 @@ class RESTResponse(io.IOBase):
"""
return self.urllib3_response.getheader(name, default)
class RESTClientObject(object):
def __init__(self, pools_size=4):
@ -87,14 +88,17 @@ class RESTClientObject(object):
:param query_params: query parameters in the url
:param headers: http request headers
:param body: request json body, for `application/json`
:param post_params: request post parameters, `application/x-www-form-urlencode`
:param post_params: request post parameters,
`application/x-www-form-urlencode`
and `multipart/form-data`
"""
method = method.upper()
assert method in ['GET', 'HEAD', 'DELETE', 'POST', 'PUT', 'PATCH']
if post_params and body:
raise ValueError("body parameter cannot be used with post_params parameter.")
raise ValueError(
"body parameter cannot be used with post_params parameter."
)
post_params = post_params or {}
headers = headers or {}
@ -144,22 +148,37 @@ class RESTClientObject(object):
return r
def GET(self, url, headers=None, query_params=None):
return self.request("GET", url, headers=headers, query_params=query_params)
return self.request("GET", url,
headers=headers,
query_params=query_params)
def HEAD(self, url, headers=None, query_params=None):
return self.request("HEAD", url, headers=headers, query_params=query_params)
return self.request("HEAD", url,
headers=headers,
query_params=query_params)
def DELETE(self, url, headers=None, query_params=None):
return self.request("DELETE", url, headers=headers, query_params=query_params)
return self.request("DELETE", url,
headers=headers,
query_params=query_params)
def POST(self, url, headers=None, post_params=None, body=None):
return self.request("POST", url, headers=headers, post_params=post_params, body=body)
return self.request("POST", url,
headers=headers,
post_params=post_params,
body=body)
def PUT(self, url, headers=None, post_params=None, body=None):
return self.request("PUT", url, headers=headers, post_params=post_params, body=body)
return self.request("PUT", url,
headers=headers,
post_params=post_params,
body=body)
def PATCH(self, url, headers=None, post_params=None, body=None):
return self.request("PATCH", url, headers=headers, post_params=post_params, body=body)
return self.request("PATCH", url,
headers=headers,
post_params=post_params,
body=body)
class ApiException(Exception):
@ -190,6 +209,7 @@ class ApiException(Exception):
return error_message
class RESTClient(object):
"""
A class with all class methods to perform JSON requests.

View File

@ -27,415 +27,480 @@ from datetime import date
from six import iteritems
try:
# for python3
from urllib.parse import quote
# for python3
from urllib.parse import quote
except ImportError:
# for python2
from urllib import quote
# for python2
from urllib import quote
from .configuration import Configuration
class ApiClient(object):
"""
Generic API client for Swagger client library builds
:param host: The base path for the server to call
:param header_name: a header to pass when making calls to the API
:param header_value: a header value to pass when making calls to the API
"""
def __init__(self, host=Configuration().host, header_name=None, header_value=None):
self.default_headers = {}
if header_name is not None:
self.default_headers[header_name] = header_value
self.host = host
self.cookie = None
# Set default User-Agent.
self.user_agent = 'Python-Swagger'
@property
def user_agent(self):
return self.default_headers['User-Agent']
@user_agent.setter
def user_agent(self, value):
self.default_headers['User-Agent'] = value
def set_default_header(self, header_name, header_value):
self.default_headers[header_name] = header_value
def __call_api(self, resource_path, method, path_params=None, query_params=None, header_params=None,
body=None, post_params=None, files=None, response_type=None, auth_settings=None, callback=None):
# headers parameters
header_params = header_params or {}
header_params.update(self.default_headers)
if self.cookie:
header_params['Cookie'] = self.cookie
if header_params:
header_params = self.sanitize_for_serialization(header_params)
# path parameters
if path_params:
path_params = self.sanitize_for_serialization(path_params)
for k, v in iteritems(path_params):
replacement = quote(str(self.to_path_value(v)))
resource_path = resource_path.replace('{' + k + '}', replacement)
# query parameters
if query_params:
query_params = self.sanitize_for_serialization(query_params)
query_params = {k: self.to_path_value(v) for k, v in iteritems(query_params)}
# post parameters
if post_params:
post_params = self.prepare_post_parameters(post_params, files)
post_params = self.sanitize_for_serialization(post_params)
# auth setting
self.update_params_for_auth(header_params, query_params, auth_settings)
# body
if body:
body = self.sanitize_for_serialization(body)
# request url
url = self.host + resource_path
# perform request and return response
response_data = self.request(method, url, query_params=query_params, headers=header_params,
post_params=post_params, body=body)
self.last_response = response_data
# deserialize response data
if response_type:
deserialized_data = self.deserialize(response_data, response_type)
else:
deserialized_data = None
if callback:
callback(deserialized_data)
else:
return deserialized_data
def to_path_value(self, obj):
"""
Convert a string or object to a path-friendly value
Generic API client for Swagger client library builds
:param obj: object or string value
:return string: quoted value
:param host: The base path for the server to call
:param header_name: a header to pass when making calls to the API
:param header_value: a header value to pass when making calls to the API
"""
if type(obj) == list:
return ','.join(obj)
else:
return str(obj)
def __init__(self, host=Configuration().host,
header_name=None, header_value=None):
def sanitize_for_serialization(self, obj):
"""
Sanitize an object for Request.
self.default_headers = {}
if header_name is not None:
self.default_headers[header_name] = header_value
self.host = host
self.cookie = None
# Set default User-Agent.
self.user_agent = 'Python-Swagger'
If obj is None, return None.
If obj is str, int, float, bool, return directly.
If obj is datetime.datetime, datetime.date convert to string in iso8601 format.
If obj is list, santize each element in the list.
If obj is dict, return the dict.
If obj is swagger model, return the properties dict.
"""
if isinstance(obj, type(None)):
return None
elif isinstance(obj, (str, int, float, bool, tuple)):
return obj
elif isinstance(obj, list):
return [self.sanitize_for_serialization(sub_obj) for sub_obj in obj]
elif isinstance(obj, (datetime, date)):
return obj.isoformat()
else:
if isinstance(obj, dict):
obj_dict = obj
else:
# Convert model obj to dict except attributes `swagger_types`, `attribute_map`
# and attributes which value is not None.
# Convert attribute name to json key in model definition for request.
obj_dict = {obj.attribute_map[key]: val
for key, val in iteritems(obj.__dict__)
if key != 'swagger_types' and key != 'attribute_map' and val is not None}
return {key: self.sanitize_for_serialization(val)
for key, val in iteritems(obj_dict)}
@property
def user_agent(self):
return self.default_headers['User-Agent']
def deserialize(self, response, response_type):
"""
Derialize response into an object.
@user_agent.setter
def user_agent(self, value):
self.default_headers['User-Agent'] = value
:param response: RESTResponse object to be deserialized
:param response_type: class literal for deserialzied object, or string of class name
def set_default_header(self, header_name, header_value):
self.default_headers[header_name] = header_value
:return: deserialized object
"""
# handle file downloading - save response body into a tmp file and return the instance
if "file" == response_type:
return self.__deserialize_file(response)
def __call_api(self, resource_path, method,
path_params=None, query_params=None, header_params=None,
body=None, post_params=None, files=None,
response_type=None, auth_settings=None, callback=None):
# fetch data from response object
try:
data = json.loads(response.data)
except ValueError:
data = response.data
# headers parameters
header_params = header_params or {}
header_params.update(self.default_headers)
if self.cookie:
header_params['Cookie'] = self.cookie
if header_params:
header_params = self.sanitize_for_serialization(header_params)
return self.__deserialize(data, response_type)
# path parameters
if path_params:
path_params = self.sanitize_for_serialization(path_params)
for k, v in iteritems(path_params):
replacement = quote(str(self.to_path_value(v)))
resource_path = resource_path.\
replace('{' + k + '}', replacement)
def __deserialize(self, data, klass):
"""
:param data: dict, list or str
:param klass: class literal, or string of class name
# query parameters
if query_params:
query_params = self.sanitize_for_serialization(query_params)
query_params = {k: self.to_path_value(v)
for k, v in iteritems(query_params)}
:return: object
"""
if data is None:
return None
# post parameters
if post_params:
post_params = self.prepare_post_parameters(post_params, files)
post_params = self.sanitize_for_serialization(post_params)
if type(klass) == str:
if 'list[' in klass:
sub_kls = re.match('list\[(.*)\]', klass).group(1)
return [self.__deserialize(sub_data, sub_kls) for sub_data in data]
# auth setting
self.update_params_for_auth(header_params, query_params, auth_settings)
if 'dict(' in klass:
sub_kls = re.match('dict\((.*), (.*)\)', klass).group(2)
return {k: self.__deserialize(v, sub_kls) for k, v in iteritems(data)}
# body
if body:
body = self.sanitize_for_serialization(body)
# convert str to class
# for native types
if klass in ['int', 'float', 'str', 'bool', "date", 'datetime', "object"]:
klass = eval(klass)
# for model types
else:
klass = eval('models.' + klass)
# request url
url = self.host + resource_path
if klass in [int, float, str, bool]:
return self.__deserialize_primitive(data, klass)
elif klass == object:
return self.__deserialize_object()
elif klass == date:
return self.__deserialize_date(data)
elif klass == datetime:
return self.__deserialize_datatime(data)
else:
return self.__deserialize_model(data, klass)
# perform request and return response
response_data = self.request(method, url,
query_params=query_params,
headers=header_params,
post_params=post_params, body=body)
def call_api(self, resource_path, method,
path_params=None, query_params=None, header_params=None,
body=None, post_params=None, files=None,
response_type=None, auth_settings=None, callback=None):
"""
Perform http request and return deserialized data
self.last_response = response_data
:param resource_path: Path to method endpoint.
:param method: Method to call.
:param path_params: Path parameters in the url.
:param query_params: Query parameters in the url.
:param header_params: Header parameters to be placed in the request header.
:param body: Request body.
:param post_params dict: Request post form parameters, for `application/x-www-form-urlencoded`, `multipart/form-data`.
:param auth_settings list: Auth Settings names for the request.
:param response: Response data type.
:param files dict: key -> filename, value -> filepath, for `multipart/form-data`.
:param callback function: Callback function for asynchronous request.
If provide this parameter, the request will be called asynchronously.
:return:
If provide parameter callback, the request will be called asynchronously.
The method will return the request thread.
If parameter callback is None, then the method will return the response directly.
"""
if callback is None:
return self.__call_api(resource_path, method,
path_params, query_params, header_params,
body, post_params, files,
response_type, auth_settings, callback)
else:
thread = threading.Thread(target=self.__call_api,
args=(resource_path, method,
path_params, query_params, header_params,
body, post_params, files,
response_type, auth_settings, callback))
thread.start()
return thread
def request(self, method, url, query_params=None, headers=None,
post_params=None, body=None):
"""
Perform http request using RESTClient.
"""
if method == "GET":
return RESTClient.GET(url, query_params=query_params, headers=headers)
elif method == "HEAD":
return RESTClient.HEAD(url, query_params=query_params, headers=headers)
elif method == "POST":
return RESTClient.POST(url, headers=headers, post_params=post_params, body=body)
elif method == "PUT":
return RESTClient.PUT(url, headers=headers, post_params=post_params, body=body)
elif method == "PATCH":
return RESTClient.PATCH(url, headers=headers, post_params=post_params, body=body)
elif method == "DELETE":
return RESTClient.DELETE(url, query_params=query_params, headers=headers)
else:
raise ValueError("http method must be `GET`, `HEAD`, `POST`, `PATCH`, `PUT` or `DELETE`")
def prepare_post_parameters(self, post_params=None, files=None):
params = {}
if post_params:
params.update(post_params)
if files:
for k, v in iteritems(files):
if v:
with open(v, 'rb') as f:
filename = os.path.basename(f.name)
filedata = f.read()
mimetype = mimetypes.guess_type(filename)[0] or 'application/octet-stream'
params[k] = tuple([filename, filedata, mimetype])
return params
def select_header_accept(self, accepts):
"""
Return `Accept` based on an array of accepts provided
"""
if not accepts:
return
accepts = list(map(lambda x: x.lower(), accepts))
if 'application/json' in accepts:
return 'application/json'
else:
return ', '.join(accepts)
def select_header_content_type(self, content_types):
"""
Return `Content-Type` baseed on an array of content_types provided
"""
if not content_types:
return 'application/json'
content_types = list(map(lambda x: x.lower(), content_types))
if 'application/json' in content_types:
return 'application/json'
else:
return content_types[0]
def update_params_for_auth(self, headers, querys, auth_settings):
"""
Update header and query params based on authentication setting
"""
config = Configuration()
if not auth_settings:
return
for auth in auth_settings:
auth_setting = config.auth_settings().get(auth)
if auth_setting:
if auth_setting['in'] == 'header':
headers[auth_setting['key']] = auth_setting['value']
elif auth_setting['in'] == 'query':
querys[auth_setting['key']] = auth_setting['value']
# deserialize response data
if response_type:
deserialized_data = self.deserialize(response_data, response_type)
else:
raise ValueError('Authentication token must be in `query` or `header`')
deserialized_data = None
def __deserialize_file(self, response):
"""
Save response body into a file in (the defined) temporary folder, using the filename
from the `Content-Disposition` header if provided, otherwise a random filename.
if callback:
callback(deserialized_data)
else:
return deserialized_data
:param response: RESTResponse
:return: file path
"""
fd, path = tempfile.mkstemp(dir=configuration.temp_folder_path)
os.close(fd)
os.remove(path)
def to_path_value(self, obj):
"""
Convert a string or object to a path-friendly value
content_disposition = response.getheader("Content-Disposition")
if content_disposition:
filename = re.search(r'filename=[\'"]?([^\'"\s]+)[\'"]?', content_disposition).group(1)
path = os.path.join(os.path.dirname(path), filename)
:param obj: object or string value
with open(path, "w") as f:
f.write(response.data)
:return string: quoted value
"""
if type(obj) == list:
return ','.join(obj)
else:
return str(obj)
return path
def sanitize_for_serialization(self, obj):
"""
Sanitize an object for Request.
def __deserialize_primitive(self, data, klass):
"""
Deserialize string to primitive type
If obj is None, return None.
If obj is str, int, float, bool, return directly.
If obj is datetime.datetime, datetime.date
convert to string in iso8601 format.
If obj is list, santize each element in the list.
If obj is dict, return the dict.
If obj is swagger model, return the properties dict.
"""
if isinstance(obj, type(None)):
return None
elif isinstance(obj, (str, int, float, bool, tuple)):
return obj
elif isinstance(obj, list):
return [self.sanitize_for_serialization(sub_obj)
for sub_obj in obj]
elif isinstance(obj, (datetime, date)):
return obj.isoformat()
else:
if isinstance(obj, dict):
obj_dict = obj
else:
# Convert model obj to dict except
# attributes `swagger_types`, `attribute_map`
# and attributes which value is not None.
# Convert attribute name to json key in
# model definition for request.
obj_dict = {obj.attribute_map[key[1:]]: val
for key, val in iteritems(obj.__dict__)
if key != 'swagger_types'
and key != 'attribute_map'
and val is not None}
:param data: str
:param klass: class literal
return {key: self.sanitize_for_serialization(val)
for key, val in iteritems(obj_dict)}
:return: int, float, str, bool
"""
try:
value = klass(data)
except UnicodeEncodeError:
value = unicode(data)
except TypeError:
value = data
return value
def deserialize(self, response, response_type):
"""
Derialize response into an object.
def __deserialize_object(self):
"""
Deserialize empty object
"""
return object()
:param response: RESTResponse object to be deserialized
:param response_type: class literal for
deserialzied object, or string of class name
def __deserialize_date(self, string):
"""
Deserialize string to date
:return: deserialized object
"""
# handle file downloading
# save response body into a tmp file and return the instance
if "file" == response_type:
return self.__deserialize_file(response)
:param string: str
:return: date
"""
try:
from dateutil.parser import parse
return parse(string).date()
except ImportError:
return string
except ValueError:
raise ApiException(status=0, reason="Failed to parse `{0}` into a date object".format(string))
# fetch data from response object
try:
data = json.loads(response.data)
except ValueError:
data = response.data
def __deserialize_datatime(self, string):
"""
Deserialize string to datetime.
return self.__deserialize(data, response_type)
The string should be in iso8601 datetime format.
def __deserialize(self, data, klass):
"""
:param data: dict, list or str
:param klass: class literal, or string of class name
:param string: str
:return: datetime
"""
try:
from dateutil.parser import parse
return parse(string)
except ImportError:
return string
except ValueError:
raise ApiException(status=0, reason="Failed to parse `{0}` into a datetime object".format(string))
:return: object
"""
if data is None:
return None
def __deserialize_model(self, data, klass):
"""
Deserialize list or dict to model
if type(klass) == str:
if 'list[' in klass:
sub_kls = re.match('list\[(.*)\]', klass).group(1)
return [self.__deserialize(sub_data, sub_kls)
for sub_data in data]
:param data: dict, list
:param klass: class literal
"""
instance = klass()
if 'dict(' in klass:
sub_kls = re.match('dict\((.*), (.*)\)', klass).group(2)
return {k: self.__deserialize(v, sub_kls)
for k, v in iteritems(data)}
for attr, attr_type in iteritems(instance.swagger_types):
if data is not None \
and instance.attribute_map[attr] in data\
and isinstance(data, (list, dict)):
value = data[instance.attribute_map[attr]]
setattr(instance, attr, self.__deserialize(value, attr_type))
# convert str to class
# for native types
if klass in ['int', 'float', 'str', 'bool',
"date", 'datetime', "object"]:
klass = eval(klass)
# for model types
else:
klass = eval('models.' + klass)
return instance
if klass in [int, float, str, bool]:
return self.__deserialize_primitive(data, klass)
elif klass == object:
return self.__deserialize_object()
elif klass == date:
return self.__deserialize_date(data)
elif klass == datetime:
return self.__deserialize_datatime(data)
else:
return self.__deserialize_model(data, klass)
def call_api(self, resource_path, method,
path_params=None, query_params=None, header_params=None,
body=None, post_params=None, files=None,
response_type=None, auth_settings=None, callback=None):
"""
Perform http request and return deserialized data
:param resource_path: Path to method endpoint.
:param method: Method to call.
:param path_params: Path parameters in the url.
:param query_params: Query parameters in the url.
:param header_params: Header parameters to be
placed in the request header.
:param body: Request body.
:param post_params dict: Request post form parameters,
for `application/x-www-form-urlencoded`, `multipart/form-data`.
:param auth_settings list: Auth Settings names for the request.
:param response: Response data type.
:param files dict: key -> filename, value -> filepath,
for `multipart/form-data`.
:param callback function: Callback function for asynchronous request.
If provide this parameter,
the request will be called asynchronously.
:return:
If provide parameter callback,
the request will be called asynchronously.
The method will return the request thread.
If parameter callback is None,
then the method will return the response directly.
"""
if callback is None:
return self.__call_api(resource_path, method,
path_params, query_params, header_params,
body, post_params, files,
response_type, auth_settings, callback)
else:
thread = threading.Thread(target=self.__call_api,
args=(resource_path, method,
path_params, query_params,
header_params, body,
post_params, files,
response_type, auth_settings,
callback))
thread.start()
return thread
def request(self, method, url, query_params=None, headers=None,
post_params=None, body=None):
"""
Perform http request using RESTClient.
"""
if method == "GET":
return RESTClient.GET(url,
query_params=query_params,
headers=headers)
elif method == "HEAD":
return RESTClient.HEAD(url,
query_params=query_params,
headers=headers)
elif method == "POST":
return RESTClient.POST(url,
headers=headers,
post_params=post_params,
body=body)
elif method == "PUT":
return RESTClient.PUT(url,
headers=headers,
post_params=post_params,
body=body)
elif method == "PATCH":
return RESTClient.PATCH(url,
headers=headers,
post_params=post_params,
body=body)
elif method == "DELETE":
return RESTClient.DELETE(url,
query_params=query_params,
headers=headers)
else:
raise ValueError(
"http method must be `GET`, `HEAD`,"
" `POST`, `PATCH`, `PUT` or `DELETE`."
)
def prepare_post_parameters(self, post_params=None, files=None):
params = {}
if post_params:
params.update(post_params)
if files:
for k, v in iteritems(files):
if not v:
continue
with open(v, 'rb') as f:
filename = os.path.basename(f.name)
filedata = f.read()
mimetype = mimetypes.\
guess_type(filename)[0] or 'application/octet-stream'
params[k] = tuple([filename, filedata, mimetype])
return params
def select_header_accept(self, accepts):
"""
Return `Accept` based on an array of accepts provided
"""
if not accepts:
return
accepts = list(map(lambda x: x.lower(), accepts))
if 'application/json' in accepts:
return 'application/json'
else:
return ', '.join(accepts)
def select_header_content_type(self, content_types):
"""
Return `Content-Type` baseed on an array of content_types provided
"""
if not content_types:
return 'application/json'
content_types = list(map(lambda x: x.lower(), content_types))
if 'application/json' in content_types:
return 'application/json'
else:
return content_types[0]
def update_params_for_auth(self, headers, querys, auth_settings):
"""
Update header and query params based on authentication setting
"""
config = Configuration()
if not auth_settings:
return
for auth in auth_settings:
auth_setting = config.auth_settings().get(auth)
if auth_setting:
if auth_setting['in'] == 'header':
headers[auth_setting['key']] = auth_setting['value']
elif auth_setting['in'] == 'query':
querys[auth_setting['key']] = auth_setting['value']
else:
raise ValueError(
'Authentication token must be in `query` or `header`'
)
def __deserialize_file(self, response):
"""
Save response body into a file in (the defined) temporary folder,
using the filename from the `Content-Disposition` header if provided,
otherwise a random filename.
:param response: RESTResponse
:return: file path
"""
config = Configuration()
fd, path = tempfile.mkstemp(dir=config.temp_folder_path)
os.close(fd)
os.remove(path)
content_disposition = response.getheader("Content-Disposition")
if content_disposition:
filename = re.\
search(r'filename=[\'"]?([^\'"\s]+)[\'"]?', content_disposition).\
group(1)
path = os.path.join(os.path.dirname(path), filename)
with open(path, "w") as f:
f.write(response.data)
return path
def __deserialize_primitive(self, data, klass):
"""
Deserialize string to primitive type
:param data: str
:param klass: class literal
:return: int, float, str, bool
"""
try:
value = klass(data)
except UnicodeEncodeError:
value = unicode(data)
except TypeError:
value = data
return value
def __deserialize_object(self):
"""
Deserialize empty object
"""
return object()
def __deserialize_date(self, string):
"""
Deserialize string to date
:param string: str
:return: date
"""
try:
from dateutil.parser import parse
return parse(string).date()
except ImportError:
return string
except ValueError:
raise ApiException(
status=0,
reason="Failed to parse `{0}` into a date object"
.format(string)
)
def __deserialize_datatime(self, string):
"""
Deserialize string to datetime.
The string should be in iso8601 datetime format.
:param string: str
:return: datetime
"""
try:
from dateutil.parser import parse
return parse(string)
except ImportError:
return string
except ValueError:
raise ApiException(
status=0,
reason="Failed to parse `{0}` into a datetime object".
format(string)
)
def __deserialize_model(self, data, klass):
"""
Deserialize list or dict to model
:param data: dict, list
:param klass: class literal
"""
instance = klass()
for attr, attr_type in iteritems(instance.swagger_types):
if data is not None \
and instance.attribute_map[attr] in data\
and isinstance(data, (list, dict)):
value = data[instance.attribute_map[attr]]
setattr(instance, attr, self.__deserialize(value, attr_type))
return instance

View File

@ -4,4 +4,3 @@ from __future__ import absolute_import
from .user_api import UserApi
from .pet_api import PetApi
from .store_api import StoreApi

View File

@ -17,7 +17,8 @@ Copyright 2015 SmartBear Software
See the License for the specific language governing permissions and
limitations under the License.
NOTE: This class is auto generated by the swagger code generator program. Do not edit the class manually.
NOTE: This class is auto generated by the swagger code generator program.
Do not edit the class manually.
"""
from __future__ import absolute_import
@ -30,6 +31,7 @@ from six import iteritems
from ..configuration import Configuration
from ..api_client import ApiClient
class PetApi(object):
def __init__(self, api_client=None):
@ -41,24 +43,25 @@ class PetApi(object):
config.api_client = ApiClient('http://petstore.swagger.io/v2')
self.api_client = config.api_client
def update_pet(self, **kwargs):
"""
Update an existing pet
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.update_pet(callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param Pet body: Pet object that needs to be added to the store
:return: None
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
all_params = ['body']
@ -67,7 +70,10 @@ class PetApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method update_pet" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method update_pet" % key
)
params[key] = val
del params['kwargs']
@ -84,24 +90,32 @@ class PetApi(object):
files = {}
body_params = None
if 'body' in params:
body_params = params['body']
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type(['application/json', 'application/xml'])
header_params['Content-Type'] = self.api_client.\
select_header_content_type(['application/json', 'application/xml'])
# Authentication setting
auth_settings = ['petstore_auth']
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type=None, auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type=None,
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def add_pet(self, **kwargs):
@ -109,18 +123,20 @@ class PetApi(object):
Add a new pet to the store
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.add_pet(callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param Pet body: Pet object that needs to be added to the store
:return: None
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
all_params = ['body']
@ -129,7 +145,10 @@ class PetApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method add_pet" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method add_pet" % key
)
params[key] = val
del params['kwargs']
@ -146,24 +165,32 @@ class PetApi(object):
files = {}
body_params = None
if 'body' in params:
body_params = params['body']
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type(['application/json', 'application/xml'])
header_params['Content-Type'] = self.api_client.\
select_header_content_type(['application/json', 'application/xml'])
# Authentication setting
auth_settings = ['petstore_auth']
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type=None, auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type=None,
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def find_pets_by_status(self, **kwargs):
@ -171,18 +198,20 @@ class PetApi(object):
Finds Pets by status
Multiple status values can be provided with comma seperated strings
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.find_pets_by_status(callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param list[str] status: Status values that need to be considered for filter
:return: list[Pet]
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
all_params = ['status']
@ -191,7 +220,10 @@ class PetApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method find_pets_by_status" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method find_pets_by_status" % key
)
params[key] = val
del params['kwargs']
@ -201,7 +233,6 @@ class PetApi(object):
path_params = {}
query_params = {}
if 'status' in params:
query_params['status'] = params['status']
@ -213,19 +244,28 @@ class PetApi(object):
body_params = None
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = ['petstore_auth']
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type='list[Pet]', auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type='list[Pet]',
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def find_pets_by_tags(self, **kwargs):
@ -233,18 +273,20 @@ class PetApi(object):
Finds Pets by tags
Muliple tags can be provided with comma seperated strings. Use tag1, tag2, tag3 for testing.
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.find_pets_by_tags(callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param list[str] tags: Tags to filter by
:return: list[Pet]
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
all_params = ['tags']
@ -253,7 +295,10 @@ class PetApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method find_pets_by_tags" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method find_pets_by_tags" % key
)
params[key] = val
del params['kwargs']
@ -263,7 +308,6 @@ class PetApi(object):
path_params = {}
query_params = {}
if 'tags' in params:
query_params['tags'] = params['tags']
@ -275,19 +319,28 @@ class PetApi(object):
body_params = None
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = ['petstore_auth']
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type='list[Pet]', auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type='list[Pet]',
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def get_pet_by_id(self, pet_id, **kwargs):
@ -295,20 +348,21 @@ class PetApi(object):
Find pet by ID
Returns a pet when ID < 10. ID > 10 or nonintegers will simulate API error conditions
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.get_pet_by_id(pet_id, callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param int pet_id: ID of pet that needs to be fetched (required)
:return: Pet
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
# verify the required parameter 'pet_id' is set
if pet_id is None:
raise ValueError("Missing the required parameter `pet_id` when calling `get_pet_by_id`")
@ -319,7 +373,10 @@ class PetApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method get_pet_by_id" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method get_pet_by_id" % key
)
params[key] = val
del params['kwargs']
@ -327,7 +384,6 @@ class PetApi(object):
method = 'GET'
path_params = {}
if 'pet_id' in params:
path_params['petId'] = params['pet_id']
@ -341,19 +397,28 @@ class PetApi(object):
body_params = None
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = ['api_key', 'petstore_auth']
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type='Pet', auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type='Pet',
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def update_pet_with_form(self, pet_id, **kwargs):
@ -361,22 +426,23 @@ class PetApi(object):
Updates a pet in the store with form data
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.update_pet_with_form(pet_id, callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param str pet_id: ID of pet that needs to be updated (required)
:param str name: Updated name of the pet
:param str status: Updated status of the pet
:return: None
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
# verify the required parameter 'pet_id' is set
if pet_id is None:
raise ValueError("Missing the required parameter `pet_id` when calling `update_pet_with_form`")
@ -387,7 +453,10 @@ class PetApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method update_pet_with_form" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method update_pet_with_form" % key
)
params[key] = val
del params['kwargs']
@ -395,7 +464,6 @@ class PetApi(object):
method = 'POST'
path_params = {}
if 'pet_id' in params:
path_params['petId'] = params['pet_id']
@ -405,29 +473,36 @@ class PetApi(object):
form_params = {}
files = {}
if 'name' in params:
form_params['name'] = params['name']
if 'status' in params:
form_params['status'] = params['status']
body_params = None
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type(['application/x-www-form-urlencoded'])
header_params['Content-Type'] = self.api_client.\
select_header_content_type(['application/x-www-form-urlencoded'])
# Authentication setting
auth_settings = ['petstore_auth']
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type=None, auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type=None,
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def delete_pet(self, pet_id, **kwargs):
@ -435,21 +510,22 @@ class PetApi(object):
Deletes a pet
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.delete_pet(pet_id, callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param int pet_id: Pet id to delete (required)
:param str api_key:
:return: None
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
# verify the required parameter 'pet_id' is set
if pet_id is None:
raise ValueError("Missing the required parameter `pet_id` when calling `delete_pet`")
@ -460,7 +536,10 @@ class PetApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method delete_pet" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method delete_pet" % key
)
params[key] = val
del params['kwargs']
@ -468,14 +547,12 @@ class PetApi(object):
method = 'DELETE'
path_params = {}
if 'pet_id' in params:
path_params['petId'] = params['pet_id']
query_params = {}
header_params = {}
if 'api_key' in params:
header_params['api_key'] = params['api_key']
@ -485,19 +562,28 @@ class PetApi(object):
body_params = None
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = ['petstore_auth']
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type=None, auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type=None,
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def upload_file(self, pet_id, **kwargs):
@ -505,22 +591,23 @@ class PetApi(object):
uploads an image
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.upload_file(pet_id, callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param int pet_id: ID of pet to update (required)
:param str additional_metadata: Additional data to pass to server
:param file file: file to upload
:return: None
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
# verify the required parameter 'pet_id' is set
if pet_id is None:
raise ValueError("Missing the required parameter `pet_id` when calling `upload_file`")
@ -531,7 +618,10 @@ class PetApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method upload_file" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method upload_file" % key
)
params[key] = val
del params['kwargs']
@ -539,7 +629,6 @@ class PetApi(object):
method = 'POST'
path_params = {}
if 'pet_id' in params:
path_params['petId'] = params['pet_id']
@ -549,28 +638,34 @@ class PetApi(object):
form_params = {}
files = {}
if 'additional_metadata' in params:
form_params['additionalMetadata'] = params['additional_metadata']
if 'file' in params:
files['file'] = params['file']
body_params = None
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type(['multipart/form-data'])
header_params['Content-Type'] = self.api_client.\
select_header_content_type(['multipart/form-data'])
# Authentication setting
auth_settings = ['petstore_auth']
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type=None, auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type=None,
auth_settings=auth_settings,
callback=params.get('callback'))
return response

View File

@ -17,7 +17,8 @@ Copyright 2015 SmartBear Software
See the License for the specific language governing permissions and
limitations under the License.
NOTE: This class is auto generated by the swagger code generator program. Do not edit the class manually.
NOTE: This class is auto generated by the swagger code generator program.
Do not edit the class manually.
"""
from __future__ import absolute_import
@ -30,6 +31,7 @@ from six import iteritems
from ..configuration import Configuration
from ..api_client import ApiClient
class StoreApi(object):
def __init__(self, api_client=None):
@ -41,23 +43,24 @@ class StoreApi(object):
config.api_client = ApiClient('http://petstore.swagger.io/v2')
self.api_client = config.api_client
def get_inventory(self, **kwargs):
"""
Returns pet inventories by status
Returns a map of status codes to quantities
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.get_inventory(callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:return: dict(str, int)
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
all_params = []
@ -66,7 +69,10 @@ class StoreApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method get_inventory" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method get_inventory" % key
)
params[key] = val
del params['kwargs']
@ -85,19 +91,28 @@ class StoreApi(object):
body_params = None
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = ['api_key']
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type='dict(str, int)', auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type='dict(str, int)',
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def place_order(self, **kwargs):
@ -105,18 +120,20 @@ class StoreApi(object):
Place an order for a pet
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.place_order(callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param Order body: order placed for purchasing the pet
:return: Order
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
all_params = ['body']
@ -125,7 +142,10 @@ class StoreApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method place_order" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method place_order" % key
)
params[key] = val
del params['kwargs']
@ -142,24 +162,32 @@ class StoreApi(object):
files = {}
body_params = None
if 'body' in params:
body_params = params['body']
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = []
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type='Order', auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type='Order',
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def get_order_by_id(self, order_id, **kwargs):
@ -167,20 +195,21 @@ class StoreApi(object):
Find purchase order by ID
For valid response try integer IDs with value <= 5 or > 10. Other values will generated exceptions
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.get_order_by_id(order_id, callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param str order_id: ID of pet that needs to be fetched (required)
:return: Order
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
# verify the required parameter 'order_id' is set
if order_id is None:
raise ValueError("Missing the required parameter `order_id` when calling `get_order_by_id`")
@ -191,7 +220,10 @@ class StoreApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method get_order_by_id" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method get_order_by_id" % key
)
params[key] = val
del params['kwargs']
@ -199,7 +231,6 @@ class StoreApi(object):
method = 'GET'
path_params = {}
if 'order_id' in params:
path_params['orderId'] = params['order_id']
@ -213,19 +244,28 @@ class StoreApi(object):
body_params = None
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = []
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type='Order', auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type='Order',
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def delete_order(self, order_id, **kwargs):
@ -233,20 +273,21 @@ class StoreApi(object):
Delete purchase order by ID
For valid response try integer IDs with value < 1000. Anything above 1000 or nonintegers will generate API errors
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.delete_order(order_id, callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param str order_id: ID of the order that needs to be deleted (required)
:return: None
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
# verify the required parameter 'order_id' is set
if order_id is None:
raise ValueError("Missing the required parameter `order_id` when calling `delete_order`")
@ -257,7 +298,10 @@ class StoreApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method delete_order" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method delete_order" % key
)
params[key] = val
del params['kwargs']
@ -265,7 +309,6 @@ class StoreApi(object):
method = 'DELETE'
path_params = {}
if 'order_id' in params:
path_params['orderId'] = params['order_id']
@ -279,18 +322,26 @@ class StoreApi(object):
body_params = None
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = []
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type=None, auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type=None,
auth_settings=auth_settings,
callback=params.get('callback'))
return response

View File

@ -17,7 +17,8 @@ Copyright 2015 SmartBear Software
See the License for the specific language governing permissions and
limitations under the License.
NOTE: This class is auto generated by the swagger code generator program. Do not edit the class manually.
NOTE: This class is auto generated by the swagger code generator program.
Do not edit the class manually.
"""
from __future__ import absolute_import
@ -30,6 +31,7 @@ from six import iteritems
from ..configuration import Configuration
from ..api_client import ApiClient
class UserApi(object):
def __init__(self, api_client=None):
@ -41,24 +43,25 @@ class UserApi(object):
config.api_client = ApiClient('http://petstore.swagger.io/v2')
self.api_client = config.api_client
def create_user(self, **kwargs):
"""
Create user
This can only be done by the logged in user.
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.create_user(callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param User body: Created user object
:return: None
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
all_params = ['body']
@ -67,7 +70,10 @@ class UserApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method create_user" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method create_user" % key
)
params[key] = val
del params['kwargs']
@ -84,24 +90,32 @@ class UserApi(object):
files = {}
body_params = None
if 'body' in params:
body_params = params['body']
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = []
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type=None, auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type=None,
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def create_users_with_array_input(self, **kwargs):
@ -109,18 +123,20 @@ class UserApi(object):
Creates list of users with given input array
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.create_users_with_array_input(callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param list[User] body: List of user object
:return: None
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
all_params = ['body']
@ -129,7 +145,10 @@ class UserApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method create_users_with_array_input" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method create_users_with_array_input" % key
)
params[key] = val
del params['kwargs']
@ -146,24 +165,32 @@ class UserApi(object):
files = {}
body_params = None
if 'body' in params:
body_params = params['body']
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = []
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type=None, auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type=None,
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def create_users_with_list_input(self, **kwargs):
@ -171,18 +198,20 @@ class UserApi(object):
Creates list of users with given input array
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.create_users_with_list_input(callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param list[User] body: List of user object
:return: None
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
all_params = ['body']
@ -191,7 +220,10 @@ class UserApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method create_users_with_list_input" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method create_users_with_list_input" % key
)
params[key] = val
del params['kwargs']
@ -208,24 +240,32 @@ class UserApi(object):
files = {}
body_params = None
if 'body' in params:
body_params = params['body']
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = []
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type=None, auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type=None,
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def login_user(self, **kwargs):
@ -233,19 +273,21 @@ class UserApi(object):
Logs user into the system
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.login_user(callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param str username: The user name for login
:param str password: The password for login in clear text
:return: str
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
all_params = ['username', 'password']
@ -254,7 +296,10 @@ class UserApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method login_user" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method login_user" % key
)
params[key] = val
del params['kwargs']
@ -264,10 +309,8 @@ class UserApi(object):
path_params = {}
query_params = {}
if 'username' in params:
query_params['username'] = params['username']
if 'password' in params:
query_params['password'] = params['password']
@ -279,19 +322,28 @@ class UserApi(object):
body_params = None
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = []
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type='str', auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type='str',
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def logout_user(self, **kwargs):
@ -299,17 +351,19 @@ class UserApi(object):
Logs out current logged in user session
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.logout_user(callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:return: None
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
all_params = []
@ -318,7 +372,10 @@ class UserApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method logout_user" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method logout_user" % key
)
params[key] = val
del params['kwargs']
@ -337,19 +394,28 @@ class UserApi(object):
body_params = None
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = []
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type=None, auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type=None,
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def get_user_by_name(self, username, **kwargs):
@ -357,20 +423,21 @@ class UserApi(object):
Get user by user name
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.get_user_by_name(username, callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param str username: The name that needs to be fetched. Use user1 for testing. (required)
:return: User
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
# verify the required parameter 'username' is set
if username is None:
raise ValueError("Missing the required parameter `username` when calling `get_user_by_name`")
@ -381,7 +448,10 @@ class UserApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method get_user_by_name" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method get_user_by_name" % key
)
params[key] = val
del params['kwargs']
@ -389,7 +459,6 @@ class UserApi(object):
method = 'GET'
path_params = {}
if 'username' in params:
path_params['username'] = params['username']
@ -403,19 +472,28 @@ class UserApi(object):
body_params = None
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = []
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type='User', auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type='User',
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def update_user(self, username, **kwargs):
@ -423,21 +501,22 @@ class UserApi(object):
Updated user
This can only be done by the logged in user.
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.update_user(username, callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param str username: name that need to be deleted (required)
:param User body: Updated user object
:return: None
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
# verify the required parameter 'username' is set
if username is None:
raise ValueError("Missing the required parameter `username` when calling `update_user`")
@ -448,7 +527,10 @@ class UserApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method update_user" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method update_user" % key
)
params[key] = val
del params['kwargs']
@ -456,7 +538,6 @@ class UserApi(object):
method = 'PUT'
path_params = {}
if 'username' in params:
path_params['username'] = params['username']
@ -468,24 +549,32 @@ class UserApi(object):
files = {}
body_params = None
if 'body' in params:
body_params = params['body']
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = []
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type=None, auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type=None,
auth_settings=auth_settings,
callback=params.get('callback'))
return response
def delete_user(self, username, **kwargs):
@ -493,20 +582,21 @@ class UserApi(object):
Delete user
This can only be done by the logged in user.
SDK also supports asynchronous requests in which you can define a `callback` function
to be passed along and invoked when receiving response:
This method makes a synchronous HTTP request by default.To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
>>> def callback_function(response):
>>> pprint(response)
>>>
>>> thread = api.delete_user(username, callback=callback_function)
:param callback function: The callback function for asynchronous request. (optional)
:param callback function: The callback function
for asynchronous request. (optional)
:param str username: The name that needs to be deleted (required)
:return: None
If the method is called asynchronously, returns the request thread.
If the method is called asynchronously,
returns the request thread.
"""
# verify the required parameter 'username' is set
if username is None:
raise ValueError("Missing the required parameter `username` when calling `delete_user`")
@ -517,7 +607,10 @@ class UserApi(object):
params = locals()
for key, val in iteritems(params['kwargs']):
if key not in all_params:
raise TypeError("Got an unexpected keyword argument '%s' to method delete_user" % key)
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method delete_user" % key
)
params[key] = val
del params['kwargs']
@ -525,7 +618,6 @@ class UserApi(object):
method = 'DELETE'
path_params = {}
if 'username' in params:
path_params['username'] = params['username']
@ -539,18 +631,26 @@ class UserApi(object):
body_params = None
# HTTP header `Accept`
header_params['Accept'] = self.api_client.select_header_accept(['application/json', 'application/xml'])
header_params['Accept'] = self.api_client.\
select_header_accept(['application/json', 'application/xml'])
if not header_params['Accept']:
del header_params['Accept']
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type([])
header_params['Content-Type'] = self.api_client.\
select_header_content_type([])
# Authentication setting
auth_settings = []
response = self.api_client.call_api(resource_path, method, path_params, query_params, header_params,
body=body_params, post_params=form_params, files=files,
response_type=None, auth_settings=auth_settings, callback=params.get('callback'))
response = self.api_client.call_api(resource_path, method,
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=files,
response_type=None,
auth_settings=auth_settings,
callback=params.get('callback'))
return response

View File

@ -5,6 +5,7 @@ import httplib
import sys
import logging
def singleton(cls, *args, **kw):
instances = {}
@ -23,6 +24,8 @@ class Configuration(object):
self.host = "http://petstore.swagger.io/v2"
# Default api client
self.api_client = None
# Temp file folder
self.temp_folder_path = None
# Authentication Settings
self.api_key = {}
self.api_key_prefix = {}
@ -94,21 +97,19 @@ class Configuration(object):
def auth_settings(self):
""" Return Auth Settings for api client """
return {
'api_key': {
'type': 'api_key',
'in': 'header',
'key': 'api_key',
'value': self.get_api_key_with_prefix('api_key')
},
}
'api_key':
{
'type': 'api_key',
'in': 'header',
'key': 'api_key',
'value': self.get_api_key_with_prefix('api_key')
},
}
def to_debug_report(self):
return "Python SDK Debug Report:\n"\
"OS: {env}\n"\
"Python Version: {pyversion}\n"\
"Version of the API: 1.0.0\n"\
"SDK Package Version: 1.0.0".format(env=sys.platform, pyversion=sys.version)
"SDK Package Version: 1.0.0".\
format(env=sys.platform, pyversion=sys.version)

View File

@ -6,4 +6,3 @@ from .category import Category
from .pet import Pet
from .tag import Tag
from .order import Order

View File

@ -16,19 +16,23 @@ Copyright 2015 SmartBear Software
See the License for the specific language governing permissions and
limitations under the License.
"""
from pprint import pformat
from six import iteritems
class Category(object):
"""
NOTE: This class is auto generated by the swagger code generator program.
Do not edit the class manually.
"""
def __init__(self):
"""
Swagger model
:param dict swaggerTypes: The key is attribute name and the value is attribute type.
:param dict attributeMap: The key is attribute name and the value is json key in definition.
:param dict swaggerTypes: The key is attribute name
and the value is attribute type.
:param dict attributeMap: The key is attribute name
and the value is json key in definition.
"""
self.swagger_types = {
'id': 'int',
@ -40,19 +44,54 @@ class Category(object):
'name': 'name'
}
self._id = None
self._name = None
self.id = None # int
@property
def id(self):
return self._id
@id.setter
def id(self, id):
self._id = id
self.name = None # str
@property
def name(self):
return self._name
@name.setter
def name(self, name):
self._name = name
def to_dict(self):
"""
Return model properties dict
"""
result = {}
for name, prop in iteritems(self.__dict__):
if name == "attribute_map" or name == "swagger_types":
continue
if isinstance(prop, list):
result[name[1:]] = list(map(
lambda x: x.to_dict() if hasattr(x, "to_dict") else x,
prop
))
elif hasattr(prop, "to_dict"):
result[name[1:]] = prop.to_dict()
else:
result[name[1:]] = prop
return result
def to_str(self):
"""
Return model properties str
"""
return pformat(self.to_dict())
def __repr__(self):
properties = []
for p in self.__dict__:
if p != 'swaggerTypes' and p != 'attributeMap':
properties.append('{prop}={val!r}'.format(prop=p, val=self.__dict__[p]))
return '<{name} {props}>'.format(name=__name__, props=' '.join(properties))
"""
For `print` and `pprint`
"""
return self.to_str()

View File

@ -16,19 +16,23 @@ Copyright 2015 SmartBear Software
See the License for the specific language governing permissions and
limitations under the License.
"""
from pprint import pformat
from six import iteritems
class Order(object):
"""
NOTE: This class is auto generated by the swagger code generator program.
Do not edit the class manually.
"""
def __init__(self):
"""
Swagger model
:param dict swaggerTypes: The key is attribute name and the value is attribute type.
:param dict attributeMap: The key is attribute name and the value is json key in definition.
:param dict swaggerTypes: The key is attribute name
and the value is attribute type.
:param dict attributeMap: The key is attribute name
and the value is json key in definition.
"""
self.swagger_types = {
'id': 'int',
@ -48,31 +52,96 @@ class Order(object):
'complete': 'complete'
}
self._id = None
self._pet_id = None
self._quantity = None
self._ship_date = None
self._status = None # Order Status
self._complete = None
self.id = None # int
@property
def id(self):
return self._id
@id.setter
def id(self, id):
self._id = id
self.pet_id = None # int
@property
def pet_id(self):
return self._pet_id
@pet_id.setter
def pet_id(self, pet_id):
self._pet_id = pet_id
self.quantity = None # int
@property
def quantity(self):
return self._quantity
@quantity.setter
def quantity(self, quantity):
self._quantity = quantity
self.ship_date = None # datetime
@property
def ship_date(self):
return self._ship_date
# Order Status
self.status = None # str
@ship_date.setter
def ship_date(self, ship_date):
self._ship_date = ship_date
@property
def status(self):
return self._status
self.complete = None # bool
@status.setter
def status(self, status):
allowed_values = ["placed", "approved", "delivered"]
if status not in allowed_values:
raise ValueError(
"Invalid value for `status`, must be one of {0}"
.format(allowed_values)
)
self._status = status
@property
def complete(self):
return self._complete
@complete.setter
def complete(self, complete):
self._complete = complete
def to_dict(self):
"""
Return model properties dict
"""
result = {}
for name, prop in iteritems(self.__dict__):
if name == "attribute_map" or name == "swagger_types":
continue
if isinstance(prop, list):
result[name[1:]] = list(map(
lambda x: x.to_dict() if hasattr(x, "to_dict") else x,
prop
))
elif hasattr(prop, "to_dict"):
result[name[1:]] = prop.to_dict()
else:
result[name[1:]] = prop
return result
def to_str(self):
"""
Return model properties str
"""
return pformat(self.to_dict())
def __repr__(self):
properties = []
for p in self.__dict__:
if p != 'swaggerTypes' and p != 'attributeMap':
properties.append('{prop}={val!r}'.format(prop=p, val=self.__dict__[p]))
return '<{name} {props}>'.format(name=__name__, props=' '.join(properties))
"""
For `print` and `pprint`
"""
return self.to_str()

View File

@ -16,19 +16,23 @@ Copyright 2015 SmartBear Software
See the License for the specific language governing permissions and
limitations under the License.
"""
from pprint import pformat
from six import iteritems
class Pet(object):
"""
NOTE: This class is auto generated by the swagger code generator program.
Do not edit the class manually.
"""
def __init__(self):
"""
Swagger model
:param dict swaggerTypes: The key is attribute name and the value is attribute type.
:param dict attributeMap: The key is attribute name and the value is json key in definition.
:param dict swaggerTypes: The key is attribute name
and the value is attribute type.
:param dict attributeMap: The key is attribute name
and the value is json key in definition.
"""
self.swagger_types = {
'id': 'int',
@ -48,31 +52,96 @@ class Pet(object):
'status': 'status'
}
self._id = None
self._category = None
self._name = None
self._photo_urls = None
self._tags = None
self._status = None # pet status in the store
self.id = None # int
@property
def id(self):
return self._id
@id.setter
def id(self, id):
self._id = id
self.category = None # Category
@property
def category(self):
return self._category
@category.setter
def category(self, category):
self._category = category
self.name = None # str
@property
def name(self):
return self._name
@name.setter
def name(self, name):
self._name = name
self.photo_urls = None # list[str]
@property
def photo_urls(self):
return self._photo_urls
@photo_urls.setter
def photo_urls(self, photo_urls):
self._photo_urls = photo_urls
self.tags = None # list[Tag]
@property
def tags(self):
return self._tags
# pet status in the store
self.status = None # str
@tags.setter
def tags(self, tags):
self._tags = tags
@property
def status(self):
return self._status
@status.setter
def status(self, status):
allowed_values = ["available", "pending", "sold"]
if status not in allowed_values:
raise ValueError(
"Invalid value for `status`, must be one of {0}"
.format(allowed_values)
)
self._status = status
def to_dict(self):
"""
Return model properties dict
"""
result = {}
for name, prop in iteritems(self.__dict__):
if name == "attribute_map" or name == "swagger_types":
continue
if isinstance(prop, list):
result[name[1:]] = list(map(
lambda x: x.to_dict() if hasattr(x, "to_dict") else x,
prop
))
elif hasattr(prop, "to_dict"):
result[name[1:]] = prop.to_dict()
else:
result[name[1:]] = prop
return result
def to_str(self):
"""
Return model properties str
"""
return pformat(self.to_dict())
def __repr__(self):
properties = []
for p in self.__dict__:
if p != 'swaggerTypes' and p != 'attributeMap':
properties.append('{prop}={val!r}'.format(prop=p, val=self.__dict__[p]))
return '<{name} {props}>'.format(name=__name__, props=' '.join(properties))
"""
For `print` and `pprint`
"""
return self.to_str()

View File

@ -16,19 +16,23 @@ Copyright 2015 SmartBear Software
See the License for the specific language governing permissions and
limitations under the License.
"""
from pprint import pformat
from six import iteritems
class Tag(object):
"""
NOTE: This class is auto generated by the swagger code generator program.
Do not edit the class manually.
"""
def __init__(self):
"""
Swagger model
:param dict swaggerTypes: The key is attribute name and the value is attribute type.
:param dict attributeMap: The key is attribute name and the value is json key in definition.
:param dict swaggerTypes: The key is attribute name
and the value is attribute type.
:param dict attributeMap: The key is attribute name
and the value is json key in definition.
"""
self.swagger_types = {
'id': 'int',
@ -40,19 +44,54 @@ class Tag(object):
'name': 'name'
}
self._id = None
self._name = None
self.id = None # int
@property
def id(self):
return self._id
@id.setter
def id(self, id):
self._id = id
self.name = None # str
@property
def name(self):
return self._name
@name.setter
def name(self, name):
self._name = name
def to_dict(self):
"""
Return model properties dict
"""
result = {}
for name, prop in iteritems(self.__dict__):
if name == "attribute_map" or name == "swagger_types":
continue
if isinstance(prop, list):
result[name[1:]] = list(map(
lambda x: x.to_dict() if hasattr(x, "to_dict") else x,
prop
))
elif hasattr(prop, "to_dict"):
result[name[1:]] = prop.to_dict()
else:
result[name[1:]] = prop
return result
def to_str(self):
"""
Return model properties str
"""
return pformat(self.to_dict())
def __repr__(self):
properties = []
for p in self.__dict__:
if p != 'swaggerTypes' and p != 'attributeMap':
properties.append('{prop}={val!r}'.format(prop=p, val=self.__dict__[p]))
return '<{name} {props}>'.format(name=__name__, props=' '.join(properties))
"""
For `print` and `pprint`
"""
return self.to_str()

View File

@ -16,19 +16,23 @@ Copyright 2015 SmartBear Software
See the License for the specific language governing permissions and
limitations under the License.
"""
from pprint import pformat
from six import iteritems
class User(object):
"""
NOTE: This class is auto generated by the swagger code generator program.
Do not edit the class manually.
"""
def __init__(self):
"""
Swagger model
:param dict swaggerTypes: The key is attribute name and the value is attribute type.
:param dict attributeMap: The key is attribute name and the value is json key in definition.
:param dict swaggerTypes: The key is attribute name
and the value is attribute type.
:param dict attributeMap: The key is attribute name
and the value is json key in definition.
"""
self.swagger_types = {
'id': 'int',
@ -52,37 +56,108 @@ class User(object):
'user_status': 'userStatus'
}
self._id = None
self._username = None
self._first_name = None
self._last_name = None
self._email = None
self._password = None
self._phone = None
self._user_status = None # User Status
self.id = None # int
@property
def id(self):
return self._id
@id.setter
def id(self, id):
self._id = id
self.username = None # str
@property
def username(self):
return self._username
@username.setter
def username(self, username):
self._username = username
self.first_name = None # str
@property
def first_name(self):
return self._first_name
@first_name.setter
def first_name(self, first_name):
self._first_name = first_name
self.last_name = None # str
@property
def last_name(self):
return self._last_name
@last_name.setter
def last_name(self, last_name):
self._last_name = last_name
self.email = None # str
@property
def email(self):
return self._email
@email.setter
def email(self, email):
self._email = email
self.password = None # str
@property
def password(self):
return self._password
@password.setter
def password(self, password):
self._password = password
self.phone = None # str
@property
def phone(self):
return self._phone
# User Status
self.user_status = None # int
@phone.setter
def phone(self, phone):
self._phone = phone
@property
def user_status(self):
return self._user_status
@user_status.setter
def user_status(self, user_status):
self._user_status = user_status
def to_dict(self):
"""
Return model properties dict
"""
result = {}
for name, prop in iteritems(self.__dict__):
if name == "attribute_map" or name == "swagger_types":
continue
if isinstance(prop, list):
result[name[1:]] = list(map(
lambda x: x.to_dict() if hasattr(x, "to_dict") else x,
prop
))
elif hasattr(prop, "to_dict"):
result[name[1:]] = prop.to_dict()
else:
result[name[1:]] = prop
return result
def to_str(self):
"""
Return model properties str
"""
return pformat(self.to_dict())
def __repr__(self):
properties = []
for p in self.__dict__:
if p != 'swaggerTypes' and p != 'attributeMap':
properties.append('{prop}={val!r}'.format(prop=p, val=self.__dict__[p]))
return '<{name} {props}>'.format(name=__name__, props=' '.join(properties))
"""
For `print` and `pprint`
"""
return self.to_str()

View File

@ -51,6 +51,7 @@ class RESTResponse(io.IOBase):
"""
return self.urllib3_response.getheader(name, default)
class RESTClientObject(object):
def __init__(self, pools_size=4):
@ -87,14 +88,17 @@ class RESTClientObject(object):
:param query_params: query parameters in the url
:param headers: http request headers
:param body: request json body, for `application/json`
:param post_params: request post parameters, `application/x-www-form-urlencode`
:param post_params: request post parameters,
`application/x-www-form-urlencode`
and `multipart/form-data`
"""
method = method.upper()
assert method in ['GET', 'HEAD', 'DELETE', 'POST', 'PUT', 'PATCH']
if post_params and body:
raise ValueError("body parameter cannot be used with post_params parameter.")
raise ValueError(
"body parameter cannot be used with post_params parameter."
)
post_params = post_params or {}
headers = headers or {}
@ -144,22 +148,37 @@ class RESTClientObject(object):
return r
def GET(self, url, headers=None, query_params=None):
return self.request("GET", url, headers=headers, query_params=query_params)
return self.request("GET", url,
headers=headers,
query_params=query_params)
def HEAD(self, url, headers=None, query_params=None):
return self.request("HEAD", url, headers=headers, query_params=query_params)
return self.request("HEAD", url,
headers=headers,
query_params=query_params)
def DELETE(self, url, headers=None, query_params=None):
return self.request("DELETE", url, headers=headers, query_params=query_params)
return self.request("DELETE", url,
headers=headers,
query_params=query_params)
def POST(self, url, headers=None, post_params=None, body=None):
return self.request("POST", url, headers=headers, post_params=post_params, body=body)
return self.request("POST", url,
headers=headers,
post_params=post_params,
body=body)
def PUT(self, url, headers=None, post_params=None, body=None):
return self.request("PUT", url, headers=headers, post_params=post_params, body=body)
return self.request("PUT", url,
headers=headers,
post_params=post_params,
body=body)
def PATCH(self, url, headers=None, post_params=None, body=None):
return self.request("PATCH", url, headers=headers, post_params=post_params, body=body)
return self.request("PATCH", url,
headers=headers,
post_params=post_params,
body=body)
class ApiException(Exception):
@ -190,6 +209,7 @@ class ApiException(Exception):
return error_message
class RESTClient(object):
"""
A class with all class methods to perform JSON requests.

View File

@ -0,0 +1,25 @@
# coding: utf-8
"""
Run the tests.
$ pip install nose (optional)
$ cd swagger_client-python
$ nosetests -v
"""
import os
import time
import unittest
import swagger_client
class OrderModelTests(unittest.TestCase):
def test_status(self):
order = swagger_client.Order()
order.status = "placed"
self.assertEqual("placed", order.status)
with self.assertRaises(ValueError):
order.status = "invalid"

View File

@ -35,7 +35,7 @@ class PetApiTests(unittest.TestCase):
self.category.name = "dog"
self.tag = swagger_client.Tag()
self.tag.id = int(time.time())
self.tag.name = "blank"
self.tag.name = "swagger-codegen-python-pet-tag"
self.pet = swagger_client.Pet()
self.pet.id = int(time.time())
self.pet.name = "hello kity"

View File

@ -0,0 +1,40 @@
# coding: utf-8
"""
Run the tests.
$ pip install nose (optional)
$ cd swagger_client-python
$ nosetests -v
"""
import os
import time
import unittest
import swagger_client
class PetModelTests(unittest.TestCase):
def setUp(self):
self.pet = swagger_client.Pet()
self.pet.name = "test name"
self.pet.id = 1
self.pet.photo_urls = ["string"]
self.pet.status = "available"
cate = swagger_client.Category()
cate.id = 1
cate.name = "dog"
self.pet.category = cate
tag = swagger_client.Tag()
tag.id = 1
self.pet.tags = [tag]
def test_to_str(self):
data = ("{'category': {'id': 1, 'name': 'dog'},\n"
" 'id': 1,\n"
" 'name': 'test name',\n"
" 'photo_urls': ['string'],\n"
" 'status': 'available',\n"
" 'tags': [{'id': 1, 'name': None}]}")
self.assertEqual(data, self.pet.to_str())