#The MIT License (MIT) #Copyright (c) 2014 Microsoft Corporation #Permission is hereby granted, free of charge, to any person obtaining a copy #of this software and associated documentation files (the "Software"), to deal #in the Software without restriction, including without limitation the rights #to use, copy, modify, merge, publish, distribute, sublicense, and/or sell #copies of the Software, and to permit persons to whom the Software is #furnished to do so, subject to the following conditions: #The above copyright notice and this permission notice shall be included in all #copies or substantial portions of the Software. #THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR #IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, #FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE #AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER #LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, #OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE #SOFTWARE. """Synchronized request in the Azure Cosmos database service. """ import json from six.moves.urllib.parse import urlparse, urlencode import six import azure.cosmos.documents as documents import azure.cosmos.errors as errors import azure.cosmos.http_constants as http_constants import azure.cosmos.retry_utility as retry_utility def _IsReadableStream(obj): """Checks whether obj is a file-like readable stream. :rtype: boolean """ if (hasattr(obj, 'read') and callable(getattr(obj, 'read'))): return True return False def _RequestBodyFromData(data): """Gets request body from data. When `data` is dict and list into unicode string; otherwise return `data` without making any change. :param (str, unicode, file-like stream object, dict, list or None) data: :rtype: str, unicode, file-like stream object, or None """ if isinstance(data, six.string_types) or _IsReadableStream(data): return data elif isinstance(data, (dict, list, tuple)): json_dumped = json.dumps(data, separators=(',',':')) if six.PY2: return json_dumped.decode('utf-8') else: return json_dumped return None def _Request(global_endpoint_manager, request, connection_policy, requests_session, path, request_options, request_body): """Makes one http request using the requests module. :param _GlobalEndpointManager global_endpoint_manager: :param dict request: contains the resourceType, operationType, endpointOverride, useWriteEndpoint, useAlternateWriteEndpoint information :param documents.ConnectionPolicy connection_policy: :param requests.Session requests_session: Session object in requests module :param str resource_url: The url for the resource :param dict request_options: :param str request_body: Unicode or None :return: tuple of (result, headers) :rtype: tuple of (dict, dict) """ is_media = request_options['path'].find('media') > -1 is_media_stream = is_media and connection_policy.MediaReadMode == documents.MediaReadMode.Streamed connection_timeout = (connection_policy.MediaRequestTimeout if is_media else connection_policy.RequestTimeout) # Every request tries to perform a refresh global_endpoint_manager.refresh_endpoint_list(None) if (request.endpoint_override): base_url = request.endpoint_override else: base_url = global_endpoint_manager.resolve_service_endpoint(request) if path: resource_url = base_url + path else: resource_url = base_url parse_result = urlparse(resource_url) # The requests library now expects header values to be strings only starting 2.11, # and will raise an error on validation if they are not, so casting all header values to strings. request_options['headers'] = { header: str(value) for header, value in request_options['headers'].items() } # We are disabling the SSL verification for local emulator(localhost/127.0.0.1) or if the user # has explicitly specified to disable SSL verification. is_ssl_enabled = (parse_result.hostname != 'localhost' and parse_result.hostname != '127.0.0.1' and not connection_policy.DisableSSLVerification) if connection_policy.SSLConfiguration: ca_certs = connection_policy.SSLConfiguration.SSLCaCerts cert_files = (connection_policy.SSLConfiguration.SSLCertFile, connection_policy.SSLConfiguration.SSLKeyFile) response = requests_session.request(request_options['method'], resource_url, data = request_body, headers = request_options['headers'], timeout = connection_timeout / 1000.0, stream = is_media_stream, verify = ca_certs, cert = cert_files) else: response = requests_session.request(request_options['method'], resource_url, data = request_body, headers = request_options['headers'], timeout = connection_timeout / 1000.0, stream = is_media_stream, # If SSL is disabled, verify = false verify = is_ssl_enabled) headers = dict(response.headers) # In case of media stream response, return the response to the user and the user # will need to handle reading the response. if is_media_stream: return (response.raw, headers) data = response.content if not six.PY2: # python 3 compatible: convert data from byte to unicode string data = data.decode('utf-8') if response.status_code >= 400: raise errors.HTTPFailure(response.status_code, data, headers) result = None if is_media: result = data else: if len(data) > 0: try: result = json.loads(data) except: raise errors.JSONParseFailure(data) return (result, headers) def SynchronizedRequest(client, request, global_endpoint_manager, connection_policy, requests_session, method, path, request_data, query_params, headers): """Performs one synchronized http request according to the parameters. :param object client: Document client instance :param dict request: :param _GlobalEndpointManager global_endpoint_manager: :param documents.ConnectionPolicy connection_policy: :param requests.Session requests_session: Session object in requests module :param str method: :param str path: :param (str, unicode, file-like stream object, dict, list or None) request_data: :param dict query_params: :param dict headers: :return: tuple of (result, headers) :rtype: tuple of (dict dict) """ request_body = None if request_data: request_body = _RequestBodyFromData(request_data) if not request_body: raise errors.UnexpectedDataType( 'parameter data must be a JSON object, string or' + ' readable stream.') request_options = {} request_options['path'] = path request_options['method'] = method if query_params: request_options['path'] += '?' + urlencode(query_params) request_options['headers'] = headers if request_body and (type(request_body) is str or type(request_body) is six.text_type): request_options['headers'][http_constants.HttpHeaders.ContentLength] = ( len(request_body)) elif request_body is None: request_options['headers'][http_constants.HttpHeaders.ContentLength] = 0 # Pass _Request function with it's parameters to retry_utility's Execute method that wraps the call with retries return retry_utility._Execute(client, global_endpoint_manager, _Request, request, connection_policy, requests_session, path, request_options, request_body)