# coding: utf-8 """ Onshape REST API The Onshape REST API consumed by all clients. # noqa: E501 The version of the OpenAPI document: 1.107 Contact: api-support@onshape.zendesk.com Generated by: https://openapi-generator.tech """ from __future__ import absolute_import import io import json import logging import re import ssl import certifi # python 2 and python 3 compatibility library import six import urllib3 # Ethan added from onshape_client.apikey_headers import add_onshape_specific_headers from onshape_client.oas.exceptions import ApiException, ApiValueError from six.moves.urllib.parse import urlencode, urlparse, parse_qs logger = logging.getLogger(__name__) class RESTResponse(io.IOBase): def __init__(self, resp): self.urllib3_response = resp self.status = resp.status self.reason = resp.reason self.data = resp.data def getheaders(self): """Returns a dictionary of the response headers.""" return self.urllib3_response.getheaders() def getheader(self, name, default=None): """Returns a given response header.""" return self.urllib3_response.getheader(name, default) class RESTClientObject(object): def __init__(self, configuration, pools_size=4, maxsize=None): # urllib3.PoolManager will pass all kw parameters to connectionpool # https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/poolmanager.py#L75 # noqa: E501 # https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/connectionpool.py#L680 # noqa: E501 # maxsize is the number of requests to host that are allowed in parallel # noqa: E501 # Custom SSL certificates and client certificates: http://urllib3.readthedocs.io/en/latest/advanced-usage.html # noqa: E501 # cert_reqs if configuration.verify_ssl: cert_reqs = ssl.CERT_REQUIRED else: cert_reqs = ssl.CERT_NONE # ca_certs 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() addition_pool_args = {} if configuration.assert_hostname is not None: addition_pool_args[ "assert_hostname" ] = configuration.assert_hostname # noqa: E501 if configuration.retries is not None: addition_pool_args["retries"] = configuration.retries if maxsize is None: if configuration.connection_pool_maxsize is not None: maxsize = configuration.connection_pool_maxsize else: maxsize = 4 # https pool manager if configuration.proxy: self.pool_manager = urllib3.ProxyManager( num_pools=pools_size, maxsize=maxsize, cert_reqs=cert_reqs, ca_certs=ca_certs, cert_file=configuration.cert_file, key_file=configuration.key_file, proxy_url=configuration.proxy, proxy_headers=configuration.proxy_headers, **addition_pool_args ) else: self.pool_manager = urllib3.PoolManager( num_pools=pools_size, maxsize=maxsize, cert_reqs=cert_reqs, ca_certs=ca_certs, cert_file=configuration.cert_file, key_file=configuration.key_file, **addition_pool_args ) # Ethan added self.configuration = configuration def request( self, method, url, query_params=None, headers=None, body=None, post_params=None, _preload_content=True, _request_timeout=None, ): """Perform requests. :param method: http request method :param url: http request url :param query_params: query parameters in the url :param headers: http request headers :param body: request json body, for `application/json` :param post_params: request post parameters, `application/x-www-form-urlencoded` and `multipart/form-data` :param _preload_content: if False, the urllib3.HTTPResponse object will be returned without reading/decoding response data. Default is True. :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. """ method = method.upper() assert method in ["GET", "HEAD", "DELETE", "POST", "PUT", "PATCH", "OPTIONS"] if post_params and body: raise ApiValueError( "body parameter cannot be used with post_params parameter." ) post_params = post_params or {} headers = headers or {} timeout = None if _request_timeout: if isinstance( _request_timeout, (int,) if six.PY3 else (int, long) ): # noqa: E501,F821 timeout = urllib3.Timeout(total=_request_timeout) elif isinstance(_request_timeout, tuple) and len(_request_timeout) == 2: timeout = urllib3.Timeout( connect=_request_timeout[0], read=_request_timeout[1] ) if "Content-Type" not in headers: headers["Content-Type"] = "application/json" # Ethan added: headers, multipart_boundary = add_onshape_specific_headers( method, url, self.configuration, query_params=query_params, headers=headers ) try: # For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE` if method in ["POST", "PUT", "PATCH", "OPTIONS", "DELETE"]: if query_params: url += "?" + urlencode(query_params) if re.search("json", headers["Content-Type"], re.IGNORECASE): request_body = None if body is not None: request_body = json.dumps(body) r = self.pool_manager.request( method, url, body=request_body, preload_content=_preload_content, timeout=timeout, headers=headers, ) elif ( headers["Content-Type"] == "application/x-www-form-urlencoded" ): # noqa: E501 r = self.pool_manager.request( method, url, fields=post_params, encode_multipart=False, preload_content=_preload_content, timeout=timeout, headers=headers, ) elif "multipart/form-data" in headers["Content-Type"]: # must del headers['Content-Type'], or the correct # Content-Type which is generated by urllib3 will be # overwritten. # del headers['Content-Type'] r = self.pool_manager.request( method, url, fields=post_params, encode_multipart=True, preload_content=_preload_content, timeout=timeout, headers=headers, multipart_boundary=multipart_boundary, ) # Pass a `string` parameter directly in the body to support # other content types than Json when `body` argument is # provided in serialized form elif isinstance(body, str) or isinstance(body, bytes): request_body = body r = self.pool_manager.request( method, url, body=request_body, preload_content=_preload_content, timeout=timeout, headers=headers, ) else: # Cannot generate the request from given parameters msg = """Cannot prepare a request message for provided arguments. Please check that your arguments match declared content type.""" raise ApiException(status=0, reason=msg) # For `GET`, `HEAD` else: r = self.pool_manager.request( method, url, fields=query_params, preload_content=_preload_content, timeout=timeout, headers=headers, redirect=False, ) except urllib3.exceptions.SSLError as e: msg = "{0}\n{1}".format(type(e).__name__, str(e)) raise ApiException(status=0, reason=msg) if _preload_content: r = RESTResponse(r) # In the python 3, the response.data is bytes. # we need to decode it to string. if six.PY3: r.data = r.data.decode("utf8") # log response body logger.debug("response body: %s", r.data) # TODO: Handle redirections # Ethan added the below clause to handle redirects correctly: if 300 <= r.status <= 399: # parse location location_string = r.getheaders()["Location"] location = urlparse(location_string) new_url = location.scheme + "://" + location.netloc + location.path logger.debug("request redirected to: " + location_string) parsed_qs = parse_qs(location.query) for q in parsed_qs: parsed_qs[q] = parsed_qs[q][0] return self.request( method, new_url, headers=headers, query_params=parsed_qs, body=body, post_params=post_params, _preload_content=_preload_content, _request_timeout=_request_timeout, ) if r.status == 403 or r.status == 401: from onshape_client.client import Client client = Client.get_client() if client.get_authentication_method() == "oauth": client.do_oauth_flow() headers["Authorization"] = "Bearer {}".format( client.configuration.access_token ) return self.request( method, url, query_params=query_params, headers=headers, body=body, post_params=post_params, _preload_content=_preload_content, _request_timeout=_request_timeout, ) if not 200 <= r.status <= 299: raise ApiException(http_resp=r) return r def GET( self, url, headers=None, query_params=None, _preload_content=True, _request_timeout=None, ): return self.request( "GET", url, headers=headers, _preload_content=_preload_content, _request_timeout=_request_timeout, query_params=query_params, ) def HEAD( self, url, headers=None, query_params=None, _preload_content=True, _request_timeout=None, ): return self.request( "HEAD", url, headers=headers, _preload_content=_preload_content, _request_timeout=_request_timeout, query_params=query_params, ) def OPTIONS( self, url, headers=None, query_params=None, post_params=None, body=None, _preload_content=True, _request_timeout=None, ): return self.request( "OPTIONS", url, headers=headers, query_params=query_params, post_params=post_params, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body, ) def DELETE( self, url, headers=None, query_params=None, body=None, _preload_content=True, _request_timeout=None, ): return self.request( "DELETE", url, headers=headers, query_params=query_params, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body, ) def POST( self, url, headers=None, query_params=None, post_params=None, body=None, _preload_content=True, _request_timeout=None, ): return self.request( "POST", url, headers=headers, query_params=query_params, post_params=post_params, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body, ) def PUT( self, url, headers=None, query_params=None, post_params=None, body=None, _preload_content=True, _request_timeout=None, ): return self.request( "PUT", url, headers=headers, query_params=query_params, post_params=post_params, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body, ) def PATCH( self, url, headers=None, query_params=None, post_params=None, body=None, _preload_content=True, _request_timeout=None, ): return self.request( "PATCH", url, headers=headers, query_params=query_params, post_params=post_params, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body, )