[Cosmos] Core pipeline integration

.gitignore

@@ -90,3 +90,4 @@ sdk/storage/azure-storage-blob/tests/settings_real.py
 sdk/storage/azure-storage-queue/tests/settings_real.py
 sdk/storage/azure-storage-file/tests/settings_real.py
 *.code-workspace
+sdk/cosmos/azure-cosmos/test/test_config.py

sdk/cosmos/azure-cosmos/azure/cosmos/__init__.py

@@ -37,6 +37,7 @@
 from .permission import Permission
 from .scripts import Scripts
 from .user import User
+from .version import VERSION
 
 __all__ = (
     "Container",
@@ -56,3 +57,4 @@
     "TriggerOperation",
     "TriggerType",
 )
+__version__ = VERSION

sdk/cosmos/azure-cosmos/azure/cosmos/_default_retry_policy.py

@@ -57,7 +57,7 @@ def __init__(self, *args):
     def needsRetry(self, error_code):
         if error_code in DefaultRetryPolicy.CONNECTION_ERROR_CODES:
             if self.args:
-                if (self.args[4]["method"] == "GET") or (http_constants.HttpHeaders.IsQuery in self.args[4]["headers"]):
+                if (self.args[3].method == "GET") or (http_constants.HttpHeaders.IsQuery in self.args[3].headers):
                     return True
                 return False
             return True

sdk/cosmos/azure-cosmos/azure/cosmos/_synchronized_request.py

@@ -33,19 +33,17 @@
 from . import _retry_utility
 
 
-def _IsReadableStream(obj):
+def _is_readable_stream(obj):
     """Checks whether obj is a file-like readable stream.
 
-    :rtype:
-        boolean
-
+    :rtype: boolean
     """
     if hasattr(obj, "read") and callable(getattr(obj, "read")):
         return True
     return False
 
 
-def _RequestBodyFromData(data):
+def _request_body_from_data(data):
     """Gets request body from data.
 
     When `data` is dict and list into unicode string; otherwise return `data`
@@ -57,7 +55,7 @@ def _RequestBodyFromData(data):
         str, unicode, file-like stream object, or None
 
     """
-    if isinstance(data, six.string_types) or _IsReadableStream(data):
+    if data is None or isinstance(data, six.string_types) or _is_readable_stream(data):
         return data
     if isinstance(data, (dict, list, tuple)):
 
@@ -66,57 +64,49 @@ def _RequestBodyFromData(data):
         if six.PY2:
             return json_dumped.decode("utf-8")
         return json_dumped
-
     return None
 
 
-def _Request(
-    global_endpoint_manager, request, connection_policy, requests_session, path, request_options, request_body
-):
+def _Request(global_endpoint_manager, request_params, connection_policy, pipeline_client, request, **kwargs):
     """Makes one http request using the requests module.
 
     :param _GlobalEndpointManager global_endpoint_manager:
-    :param dict request:
+    :param dict request_params:
         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
+    :param azure.core.PipelineClient pipeline_client:
+        Pipeline client to process the resquest
+    :param azure.core.HttpRequest request:
+        The request object to send through the pipeline
 
     :return:
         tuple of (result, headers)
     :rtype:
         tuple of (dict, dict)
 
     """
-    is_media = request_options["path"].find("media") > -1
+    is_media = request.url.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
+    connection_timeout = kwargs.pop("connection_timeout", connection_timeout / 1000.0)
 
     # Every request tries to perform a refresh
     global_endpoint_manager.refresh_endpoint_list(None)
 
-    if request.endpoint_override:
-        base_url = request.endpoint_override
+    if request_params.endpoint_override:
+        base_url = request_params.endpoint_override
     else:
-        base_url = global_endpoint_manager.resolve_service_endpoint(request)
+        base_url = global_endpoint_manager.resolve_service_endpoint(request_params)
+    if base_url != pipeline_client._base_url:
+        request.url = request.url.replace(pipeline_client._base_url, base_url)
 
-    if path:
-        resource_url = base_url + path
-    else:
-        resource_url = base_url
-
-    parse_result = urlparse(resource_url)
+    parse_result = urlparse(request.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()}
+    request.headers.update({header: str(value) for header, value in request.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.
@@ -126,40 +116,35 @@ def _Request(
         and not connection_policy.DisableSSLVerification
     )
 
-    if connection_policy.SSLConfiguration:
+    if connection_policy.SSLConfiguration or "connection_cert" in kwargs:
         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,
+        response = pipeline_client._pipeline.run(
+            request,
             stream=is_media_stream,
-            verify=ca_certs,
-            cert=cert_files,
+            connection_timeout=connection_timeout,
+            connection_verify=kwargs.pop("connection_verify", ca_certs),
+            connection_cert=kwargs.pop("connection_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,
+        response = pipeline_client._pipeline.run(
+            request,
             stream=is_media_stream,
+            connection_timeout=connection_timeout,
             # If SSL is disabled, verify = false
-            verify=is_ssl_enabled,
+            connection_verify=kwargs.pop("connection_verify", is_ssl_enabled)
         )
 
+    response = response.http_response
     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)
+        return (response.stream_download(pipeline_client._pipeline), headers)
 
-    data = response.content
+    data = response.body()
     if not six.PY2:
         # python 3 compatible: convert data from byte to unicode string
         data = data.decode("utf-8")
@@ -182,25 +167,23 @@ def _Request(
 
 def SynchronizedRequest(
     client,
-    request,
+    request_params,
     global_endpoint_manager,
     connection_policy,
-    requests_session,
-    method,
-    path,
+    pipeline_client,
+    request,
     request_data,
-    query_params,
-    headers,
+    **kwargs
 ):
     """Performs one synchronized http request according to the parameters.
 
     :param object client:
         Document client instance
-    :param dict request:
-    :param _GlobalEndpointManager global_endpoint_manager:
+    :param dict request_params:
+    :param _GlobalEndpointManager global_endpoint_manager: 
     :param  documents.ConnectionPolicy connection_policy:
-    :param requests.Session requests_session:
-        Session object in requests module
+    :param azure.core.PipelineClient pipeline_client:
+        PipelineClient to process the request.
     :param str method:
     :param str path:
     :param (str, unicode, file-like stream object, dict, list or None) request_data:
@@ -213,33 +196,20 @@ def SynchronizedRequest(
         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 isinstance(request_body, (str, 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
+    request.data = _request_body_from_data(request_data)
+    if request.data and isinstance(request.data, six.string_types):
+        request.headers[http_constants.HttpHeaders.ContentLength] = len(request.data)
+    elif request.data is None:
+        request.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,
+        request_params,
         connection_policy,
-        requests_session,
-        path,
-        request_options,
-        request_body,
+        pipeline_client,
+        request,
+        **kwargs
     )