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

Merge pull request #1132 from geekerzp/python-configuration

Browse Source 
[Python] Add option `verify_ssl` to switch SSL/TLS verification
This commit is contained in:
wing328 2015-08-27 17:01:29 +08:00
commit 9e5de7da5e
10 changed files with 180 additions and 240 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/swagger-codegen/src/main/resources/python/api.mustache
View File

@ -43,7 +43,7 @@ class {{classname}}(object):
self.api_client = api_client
else:
if not config.api_client:
config.api_client = ApiClient('{{basePath}}')
config.api_client = ApiClient()
self.api_client = config.api_client
{{#operation}}

15
modules/swagger-codegen/src/main/resources/python/api_client.mustache
View File

@ -20,7 +20,7 @@ Copyright 2015 SmartBear Software
from __future__ import absolute_import
from . import models
from .rest import RESTClient
from .rest import RESTClientObject
from .rest import ApiException
import os
@ -71,6 +71,7 @@ class ApiClient(object):
"""
Constructor of the class.
"""
self.rest_client = RESTClientObject()
self.default_headers = {}
if header_name is not None:
self.default_headers[header_name] = header_value
@ -330,33 +331,33 @@ class ApiClient(object):
Makes the HTTP request using RESTClient.
"""
if method == "GET":
return RESTClient.GET(url,
return self.rest_client.GET(url,
query_params=query_params,
headers=headers)
elif method == "HEAD":
return RESTClient.HEAD(url,
return self.rest_client.HEAD(url,
query_params=query_params,
headers=headers)
elif method == "POST":
return RESTClient.POST(url,
return self.rest_client.POST(url,
query_params=query_params,
headers=headers,
post_params=post_params,
body=body)
elif method == "PUT":
return RESTClient.PUT(url,
return self.rest_client.PUT(url,
query_params=query_params,
headers=headers,
post_params=post_params,
body=body)
elif method == "PATCH":
return RESTClient.PATCH(url,
return self.rest_client.PATCH(url,
query_params=query_params,
headers=headers,
post_params=post_params,
body=body)
elif method == "DELETE":
return RESTClient.DELETE(url,
return self.rest_client.DELETE(url,
query_params=query_params,
headers=headers)
else:

6
modules/swagger-codegen/src/main/resources/python/configuration.mustache
View File

@ -79,6 +79,12 @@ class Configuration(object):
self.__debug = False
self.init_logger()
# SSL/TLS verification
# Set this to false to skip verifying SSL certificate when calling API from https server.
self.verify_ssl = True
# Set this to customize the certificate file to verify the peer.
self.ssl_ca_cert = None
def init_logger(self):
"""
Initializes logger settings.

105
modules/swagger-codegen/src/main/resources/python/rest.mustache
View File

@ -18,6 +18,7 @@ Copyright 2015 SmartBear Software
Credit: this file (rest.py) is modified based on rest.py in Dropbox Python SDK:
https://www.dropbox.com/developers/core/sdks/python
"""
from __future__ import absolute_import
import sys
import io
@ -29,6 +30,8 @@ import logging
# python 2 and python 3 compatibility library
from six import iteritems
from .configuration import Configuration
try:
import urllib3
except ImportError:
@ -69,31 +72,24 @@ class RESTResponse(io.IOBase):
class RESTClientObject(object):
def __init__(self, pools_size=4):
# http pool manager
self.pool_manager = urllib3.PoolManager(
num_pools=pools_size
)
if Configuration().verify_ssl:
cert_reqs = ssl.CERT_REQUIRED
else:
cert_reqs = ssl.CERT_NONE
if Configuration().ssl_ca_cert:
ca_certs = Configuration().ssl_ca_cert
else:
# if not set certificate file, use Mozilla's root certificates.
ca_certs = certifi.where()
# https pool manager
# certificates validated using Mozilla’s root certificates
self.ssl_pool_manager = urllib3.PoolManager(
self.pool_manager = urllib3.PoolManager(
num_pools=pools_size,
cert_reqs=ssl.CERT_REQUIRED,
ca_certs=certifi.where()
cert_reqs=cert_reqs,
ca_certs=ca_certs
)
def agent(self, url):
"""
Use `urllib3.util.parse_url` for backward compatibility.
Return proper pool manager for the http/https schemes.
"""
url = urllib3.util.parse_url(url)
scheme = url.scheme
if scheme == 'https':
return self.ssl_pool_manager
else:
return self.pool_manager
def request(self, method, url, query_params=None, headers=None,
body=None, post_params=None):
"""
@ -120,16 +116,17 @@ class RESTClientObject(object):
if 'Content-Type' not in headers:
headers['Content-Type'] = 'application/json'
try:
# For `POST`, `PUT`, `PATCH`
if method in ['POST', 'PUT', 'PATCH']:
if query_params:
url += '?' + urlencode(query_params)
if headers['Content-Type'] == 'application/json':
r = self.agent(url).request(method, url,
r = self.pool_manager.request(method, url,
body=json.dumps(body),
headers=headers)
if headers['Content-Type'] == 'application/x-www-form-urlencoded':
r = self.agent(url).request(method, url,
r = self.pool_manager.request(method, url,
fields=post_params,
encode_multipart=False,
headers=headers)
@ -137,15 +134,19 @@ class RESTClientObject(object):
# must del headers['Content-Type'], or the correct Content-Type
# which generated by urllib3 will be overwritten.
del headers['Content-Type']
r = self.agent(url).request(method, url,
r = self.pool_manager.request(method, url,
fields=post_params,
encode_multipart=True,
headers=headers)
# For `GET`, `HEAD`, `DELETE`
else:
r = self.agent(url).request(method, url,
r = self.pool_manager.request(method, url,
fields=query_params,
headers=headers)
except urllib3.exceptions.SSLError as e:
msg = "{0}\n{1}".format(type(e).__name__, str(e))
raise ApiException(status=0, reason=msg)
r = RESTResponse(r)
# In the python 3, the response.data is bytes.
@ -227,58 +228,20 @@ class ApiException(Exception):
return error_message
class RESTClient(object):
"""
A class with all class methods to perform JSON requests.
"""
IMPL = RESTClientObject()
@classmethod
def request(cls, *n, **kw):
"""
Perform a REST request and parse the response.
"""
return cls.IMPL.request(*n, **kw)
@classmethod
def GET(cls, *n, **kw):
"""
Perform a GET request using `RESTClient.request()`.
"""
return cls.IMPL.GET(*n, **kw)
@classmethod
def HEAD(cls, *n, **kw):
"""
Perform a HEAD request using `RESTClient.request()`.
"""
return cls.IMPL.GET(*n, **kw)
@classmethod
def POST(cls, *n, **kw):
"""
Perform a POST request using `RESTClient.request()`
"""
return cls.IMPL.POST(*n, **kw)
@classmethod
def PUT(cls, *n, **kw):
"""
Perform a PUT request using `RESTClient.request()`
"""
return cls.IMPL.PUT(*n, **kw)
@classmethod
def PATCH(cls, *n, **kw):
"""
Perform a PATCH request using `RESTClient.request()`
"""
return cls.IMPL.PATCH(*n, **kw)
@classmethod
def DELETE(cls, *n, **kw):
"""
Perform a DELETE request using `RESTClient.request()`
"""
return cls.IMPL.DELETE(*n, **kw)

15
samples/client/petstore/python/swagger_client/api_client.py
View File

@ -20,7 +20,7 @@ Copyright 2015 SmartBear Software
from __future__ import absolute_import
from . import models
from .rest import RESTClient
from .rest import RESTClientObject
from .rest import ApiException
import os
@ -71,6 +71,7 @@ class ApiClient(object):
"""
Constructor of the class.
"""
self.rest_client = RESTClientObject()
self.default_headers = {}
if header_name is not None:
self.default_headers[header_name] = header_value
@ -330,33 +331,33 @@ class ApiClient(object):
Makes the HTTP request using RESTClient.
"""
if method == "GET":
return RESTClient.GET(url,
return self.rest_client.GET(url,
query_params=query_params,
headers=headers)
elif method == "HEAD":
return RESTClient.HEAD(url,
return self.rest_client.HEAD(url,
query_params=query_params,
headers=headers)
elif method == "POST":
return RESTClient.POST(url,
return self.rest_client.POST(url,
query_params=query_params,
headers=headers,
post_params=post_params,
body=body)
elif method == "PUT":
return RESTClient.PUT(url,
return self.rest_client.PUT(url,
query_params=query_params,
headers=headers,
post_params=post_params,
body=body)
elif method == "PATCH":
return RESTClient.PATCH(url,
return self.rest_client.PATCH(url,
query_params=query_params,
headers=headers,
post_params=post_params,
body=body)
elif method == "DELETE":
return RESTClient.DELETE(url,
return self.rest_client.DELETE(url,
query_params=query_params,
headers=headers)
else:

2
samples/client/petstore/python/swagger_client/apis/pet_api.py
View File

@ -42,7 +42,7 @@ class PetApi(object):
self.api_client = api_client
else:
if not config.api_client:
config.api_client = ApiClient('http://petstore.swagger.io/v2')
config.api_client = ApiClient()
self.api_client = config.api_client
def update_pet(self, **kwargs):

2
samples/client/petstore/python/swagger_client/apis/store_api.py
View File

@ -42,7 +42,7 @@ class StoreApi(object):
self.api_client = api_client
else:
if not config.api_client:
config.api_client = ApiClient('http://petstore.swagger.io/v2')
config.api_client = ApiClient()
self.api_client = config.api_client
def get_inventory(self, **kwargs):

2
samples/client/petstore/python/swagger_client/apis/user_api.py
View File

@ -42,7 +42,7 @@ class UserApi(object):
self.api_client = api_client
else:
if not config.api_client:
config.api_client = ApiClient('http://petstore.swagger.io/v2')
config.api_client = ApiClient()
self.api_client = config.api_client
def create_user(self, **kwargs):

6
samples/client/petstore/python/swagger_client/configuration.py
View File

@ -79,6 +79,12 @@ class Configuration(object):
self.__debug = False
self.init_logger()
# SSL/TLS verification
# Set this to false to skip verifying SSL certificate when calling API from https server.
self.verify_ssl = True
# Set this to customize the certificate file to verify the peer.
self.ssl_ca_cert = None
def init_logger(self):
"""
Initializes logger settings.

105
samples/client/petstore/python/swagger_client/rest.py
View File

@ -18,6 +18,7 @@ Copyright 2015 SmartBear Software
Credit: this file (rest.py) is modified based on rest.py in Dropbox Python SDK:
https://www.dropbox.com/developers/core/sdks/python
"""
from __future__ import absolute_import
import sys
import io
@ -29,6 +30,8 @@ import logging
# python 2 and python 3 compatibility library
from six import iteritems
from .configuration import Configuration
try:
import urllib3
except ImportError:
@ -69,31 +72,24 @@ class RESTResponse(io.IOBase):
class RESTClientObject(object):
def __init__(self, pools_size=4):
# http pool manager
self.pool_manager = urllib3.PoolManager(
num_pools=pools_size
)
if Configuration().verify_ssl:
cert_reqs = ssl.CERT_REQUIRED
else:
cert_reqs = ssl.CERT_NONE
if Configuration().ssl_ca_cert:
ca_certs = Configuration().ssl_ca_cert
else:
# if not set certificate file, use Mozilla's root certificates.
ca_certs = certifi.where()
# https pool manager
# certificates validated using Mozilla’s root certificates
self.ssl_pool_manager = urllib3.PoolManager(
self.pool_manager = urllib3.PoolManager(
num_pools=pools_size,
cert_reqs=ssl.CERT_REQUIRED,
ca_certs=certifi.where()
cert_reqs=cert_reqs,
ca_certs=ca_certs
)
def agent(self, url):
"""
Use `urllib3.util.parse_url` for backward compatibility.
Return proper pool manager for the http/https schemes.
"""
url = urllib3.util.parse_url(url)
scheme = url.scheme
if scheme == 'https':
return self.ssl_pool_manager
else:
return self.pool_manager
def request(self, method, url, query_params=None, headers=None,
body=None, post_params=None):
"""
@ -120,16 +116,17 @@ class RESTClientObject(object):
if 'Content-Type' not in headers:
headers['Content-Type'] = 'application/json'
try:
# For `POST`, `PUT`, `PATCH`
if method in ['POST', 'PUT', 'PATCH']:
if query_params:
url += '?' + urlencode(query_params)
if headers['Content-Type'] == 'application/json':
r = self.agent(url).request(method, url,
r = self.pool_manager.request(method, url,
body=json.dumps(body),
headers=headers)
if headers['Content-Type'] == 'application/x-www-form-urlencoded':
r = self.agent(url).request(method, url,
r = self.pool_manager.request(method, url,
fields=post_params,
encode_multipart=False,
headers=headers)
@ -137,15 +134,19 @@ class RESTClientObject(object):
# must del headers['Content-Type'], or the correct Content-Type
# which generated by urllib3 will be overwritten.
del headers['Content-Type']
r = self.agent(url).request(method, url,
r = self.pool_manager.request(method, url,
fields=post_params,
encode_multipart=True,
headers=headers)
# For `GET`, `HEAD`, `DELETE`
else:
r = self.agent(url).request(method, url,
r = self.pool_manager.request(method, url,
fields=query_params,
headers=headers)
except urllib3.exceptions.SSLError as e:
msg = "{0}\n{1}".format(type(e).__name__, str(e))
raise ApiException(status=0, reason=msg)
r = RESTResponse(r)
# In the python 3, the response.data is bytes.
@ -227,58 +228,20 @@ class ApiException(Exception):
return error_message
class RESTClient(object):
"""
A class with all class methods to perform JSON requests.
"""
IMPL = RESTClientObject()
@classmethod
def request(cls, *n, **kw):
"""
Perform a REST request and parse the response.
"""
return cls.IMPL.request(*n, **kw)
@classmethod
def GET(cls, *n, **kw):
"""
Perform a GET request using `RESTClient.request()`.
"""
return cls.IMPL.GET(*n, **kw)
@classmethod
def HEAD(cls, *n, **kw):
"""
Perform a HEAD request using `RESTClient.request()`.
"""
return cls.IMPL.GET(*n, **kw)
@classmethod
def POST(cls, *n, **kw):
"""
Perform a POST request using `RESTClient.request()`
"""
return cls.IMPL.POST(*n, **kw)
@classmethod
def PUT(cls, *n, **kw):
"""
Perform a PUT request using `RESTClient.request()`
"""
return cls.IMPL.PUT(*n, **kw)
@classmethod
def PATCH(cls, *n, **kw):
"""
Perform a PATCH request using `RESTClient.request()`
"""
return cls.IMPL.PATCH(*n, **kw)
@classmethod
def DELETE(cls, *n, **kw):
"""
Perform a DELETE request using `RESTClient.request()`
"""
return cls.IMPL.DELETE(*n, **kw)