Connection management support (with optional grpc_gcp dependency)

.gitignore

@@ -45,6 +45,9 @@ htmlcov
 # JetBrains
 .idea
 
+# VS Code
+.vscode
+
 # Built documentation
 docs/_build
 docs/_build_doc2dash

api_core/google/api_core/grpc_helpers.py

@@ -26,6 +26,11 @@
 import google.auth.transport.grpc
 import google.auth.transport.requests
 
+try:
+    import grpc_gcp
+    HAS_GRPC_GCP = True
+except ImportError:
+    HAS_GRPC_GCP = False
 
 # The list of gRPC Callable interfaces that return iterators.
 _STREAM_WRAP_CLASSES = (
@@ -149,7 +154,11 @@ def wrap_errors(callable_):
         return _wrap_unary_errors(callable_)
 
 
-def create_channel(target, credentials=None, scopes=None, **kwargs):
+def create_channel(target,
+                   credentials=None,
+                   scopes=None,
+                   ssl_credentials=None,
+                   **kwargs):
     """Create a secure channel with credentials.
 
     Args:
@@ -160,8 +169,10 @@ def create_channel(target, credentials=None, scopes=None, **kwargs):
         scopes (Sequence[str]): A optional list of scopes needed for this
             service. These are only used when credentials are not specified and
             are passed to :func:`google.auth.default`.
+        ssl_credentials (grpc.ChannelCredentials): Optional SSL channel
+            credentials. This can be used to specify different certificates.
         kwargs: Additional key-word args passed to
-            :func:`google.auth.transport.grpc.secure_authorized_channel`.
+            :func:`grpc_gcp.secure_channel` or :func:`grpc.secure_channel`.
 
     Returns:
         grpc.Channel: The created channel.
@@ -174,8 +185,26 @@ def create_channel(target, credentials=None, scopes=None, **kwargs):
 
     request = google.auth.transport.requests.Request()
 
-    return google.auth.transport.grpc.secure_authorized_channel(
-        credentials, request, target, **kwargs)
+    # Create the metadata plugin for inserting the authorization header.
+    metadata_plugin = google.auth.transport.grpc.AuthMetadataPlugin(
+        credentials, request)
+
+    # Create a set of grpc.CallCredentials using the metadata plugin.
+    google_auth_credentials = grpc.metadata_call_credentials(metadata_plugin)
+
+    if ssl_credentials is None:
+        ssl_credentials = grpc.ssl_channel_credentials()
+
+    # Combine the ssl credentials and the authorization credentials.
+    composite_credentials = grpc.composite_channel_credentials(
+        ssl_credentials, google_auth_credentials)
+
+    if HAS_GRPC_GCP:
+        # If grpc_gcp module is available use grpc_gcp.secure_channel,
+        # otherwise, use grpc.secure_channel to create grpc channel.
+        return grpc_gcp.secure_channel(target, composite_credentials, **kwargs)
+    else:
+        return grpc.secure_channel(target, composite_credentials, **kwargs)
 
 
 _MethodCall = collections.namedtuple(

api_core/nox.py

@@ -65,6 +65,23 @@ def unit(session, py):
     default(session)
 
 
+@nox.session
+@nox.parametrize('py', ['2.7', '3.5', '3.6', '3.7'])
+def unit_grpc_gcp(session, py):
+    """Run the unit test suite with grpcio-gcp installed."""
+
+    # Run unit tests against all supported versions of Python.
+    session.interpreter = 'python{}'.format(py)
+
+    # Set the virtualenv dirname.
+    session.virtualenv_dirname = 'unit-grpc-gcp-' + py
+
+    # Install grpcio-gcp
+    session.install('grpcio-gcp')
+
+    default(session)
+
+
 @nox.session
 def lint(session):
     """Run linters.

api_core/tests/unit/test_grpc_helpers.py

@@ -176,60 +176,137 @@ def test_wrap_errors_streaming(wrap_stream_errors):
     wrap_stream_errors.assert_called_once_with(callable_)
 
 
+@mock.patch('grpc.composite_channel_credentials')
 @mock.patch(
     'google.auth.default',
     return_value=(mock.sentinel.credentials, mock.sentinel.projet))
-@mock.patch('google.auth.transport.grpc.secure_authorized_channel')
-def test_create_channel_implicit(secure_authorized_channel, default):
+@mock.patch('grpc._channel.Channel')
+def test_create_channel_implicit(grpc_channel, default, composite_creds_call):
     target = 'example.com:443'
+    composite_creds = composite_creds_call.return_value
+    composite_creds._credentials = mock.sentinel.channel_creds
 
     channel = grpc_helpers.create_channel(target)
 
-    assert channel is secure_authorized_channel.return_value
+    assert channel is grpc_channel.return_value
     default.assert_called_once_with(scopes=None)
-    secure_authorized_channel.assert_called_once_with(
-        mock.sentinel.credentials, mock.ANY, target)
+    grpc_channel.assert_called_once_with(
+        target, (), mock.sentinel.channel_creds)
 
 
+@mock.patch('grpc._channel.Channel')
+@mock.patch('grpc.composite_channel_credentials')
 @mock.patch(
     'google.auth.default',
     return_value=(mock.sentinel.credentials, mock.sentinel.projet))
-@mock.patch('google.auth.transport.grpc.secure_authorized_channel')
+def test_create_channel_implicit_with_ssl_creds(
+        default, composite, grpc_channel):
+    target = 'example.com:443'
+
+    ssl_creds = grpc.ssl_channel_credentials()
+
+    grpc_helpers.create_channel(target, ssl_credentials=ssl_creds)
+
+    default.assert_called_once_with(scopes=None)
+    composite.assert_called_once_with(ssl_creds, mock.ANY)
+    grpc_channel.assert_called()
+
+
+@mock.patch('grpc.composite_channel_credentials')
+@mock.patch(
+    'google.auth.default',
+    return_value=(mock.sentinel.credentials, mock.sentinel.projet))
+@mock.patch('grpc._channel.Channel')
 def test_create_channel_implicit_with_scopes(
-        secure_authorized_channel, default):
+        grpc_channel, default, composite_creds_call):
     target = 'example.com:443'
+    composite_creds = composite_creds_call.return_value
+    composite_creds._credentials = mock.sentinel.channel_creds
 
     channel = grpc_helpers.create_channel(target, scopes=['one', 'two'])
 
-    assert channel is secure_authorized_channel.return_value
+    assert channel is grpc_channel.return_value
     default.assert_called_once_with(scopes=['one', 'two'])
+    grpc_channel.assert_called_once_with(
+        target, (), mock.sentinel.channel_creds)
 
 
-@mock.patch('google.auth.transport.grpc.secure_authorized_channel')
-def test_create_channel_explicit(secure_authorized_channel):
+@mock.patch('grpc.composite_channel_credentials')
+@mock.patch('google.auth.credentials.with_scopes_if_required')
+@mock.patch('grpc._channel.Channel')
+def test_create_channel_explicit(
+        grpc_channel, auth_creds, composite_creds_call):
     target = 'example.com:443'
+    composite_creds = composite_creds_call.return_value
+    composite_creds._credentials = mock.sentinel.channel_creds
 
     channel = grpc_helpers.create_channel(
         target, credentials=mock.sentinel.credentials)
 
-    assert channel is secure_authorized_channel.return_value
-    secure_authorized_channel.assert_called_once_with(
-        mock.sentinel.credentials, mock.ANY, target)
+    auth_creds.assert_called_once_with(mock.sentinel.credentials, None)
+    assert channel is grpc_channel.return_value
+    grpc_channel.assert_called_once_with(
+        target, (), mock.sentinel.channel_creds)
 
 
-@mock.patch('google.auth.transport.grpc.secure_authorized_channel')
-def test_create_channel_explicit_scoped(unused_secure_authorized_channel):
+@mock.patch('grpc.composite_channel_credentials')
+@mock.patch('grpc._channel.Channel')
+def test_create_channel_explicit_scoped(grpc_channel, composite_creds_call):
+    target = 'example.com:443'
     scopes = ['1', '2']
+    composite_creds = composite_creds_call.return_value
+    composite_creds._credentials = mock.sentinel.channel_creds
+
+    credentials = mock.create_autospec(
+        google.auth.credentials.Scoped, instance=True)
+    credentials.requires_scopes = True
+
+    channel = grpc_helpers.create_channel(
+        target,
+        credentials=credentials,
+        scopes=scopes)
+
+    credentials.with_scopes.assert_called_once_with(scopes)
+    assert channel is grpc_channel.return_value
+    grpc_channel.assert_called_once_with(
+        target, (), mock.sentinel.channel_creds)
+
+
+@pytest.mark.skipif(not grpc_helpers.HAS_GRPC_GCP,
+                    reason='grpc_gcp module not available')
+@mock.patch('grpc_gcp.secure_channel')
+def test_create_channel_with_grpc_gcp(grpc_gcp_secure_channel):
+    target = 'example.com:443'
+    scopes = ['test_scope']
 
     credentials = mock.create_autospec(
         google.auth.credentials.Scoped, instance=True)
     credentials.requires_scopes = True
 
     grpc_helpers.create_channel(
-        mock.sentinel.target,
+        target,
         credentials=credentials,
         scopes=scopes)
+    grpc_gcp_secure_channel.assert_called()
+    credentials.with_scopes.assert_called_once_with(scopes)
 
+
+@pytest.mark.skipif(grpc_helpers.HAS_GRPC_GCP,
+                    reason='grpc_gcp module not available')
+@mock.patch('grpc.secure_channel')
+def test_create_channel_without_grpc_gcp(grpc_secure_channel):
+    target = 'example.com:443'
+    scopes = ['test_scope']
+
+    credentials = mock.create_autospec(
+        google.auth.credentials.Scoped, instance=True)
+    credentials.requires_scopes = True
+
+    grpc_helpers.create_channel(
+        target,
+        credentials=credentials,
+        scopes=scopes)
+    grpc_secure_channel.assert_called()
     credentials.with_scopes.assert_called_once_with(scopes)
 
 

spanner/MANIFEST.in

@@ -1,4 +1,4 @@
 include README.rst LICENSE
-recursive-include google *.json *.proto
+recursive-include google *.json *.proto *.config
 recursive-include tests *
 global-exclude *.pyc __pycache__

spanner/google/cloud/spanner_v1/gapic/spanner.grpc.config

@@ -0,0 +1,88 @@
+channel_pool: {
+  max_size: 10
+  max_concurrent_streams_low_watermark: 100
+}
+method: {
+  name: "/google.spanner.v1.Spanner/CreateSession"
+  affinity: {
+    command: BIND
+    affinity_key: "name"
+  }
+}
+method: {
+  name: "/google.spanner.v1.Spanner/GetSession"
+  affinity: {
+    command: BOUND
+    affinity_key: "name"
+  }
+}
+method: {
+  name: "/google.spanner.v1.Spanner/DeleteSession"
+  affinity: {
+    command: UNBIND
+    affinity_key: "name"
+  }
+}
+method: {
+  name: "/google.spanner.v1.Spanner/ExecuteSql"
+  affinity: {
+    command: BOUND
+    affinity_key: "session"
+  }
+}
+method: {
+  name: "/google.spanner.v1.Spanner/ExecuteStreamingSql"
+  affinity: {
+    command: BOUND
+    affinity_key: "session"
+  }
+}
+method: {
+  name: "/google.spanner.v1.Spanner/Read"
+  affinity: {
+    command: BOUND
+    affinity_key: "session"
+  }
+}
+method: {
+  name: "/google.spanner.v1.Spanner/StreamingRead"
+  affinity: {
+    command: BOUND
+    affinity_key: "session"
+  }
+}
+method: {
+  name: "/google.spanner.v1.Spanner/BeginTransaction"
+  affinity: {
+    command: BOUND
+    affinity_key: "session"
+  }
+}
+method: {
+  name: "/google.spanner.v1.Spanner/Commit"
+  affinity: {
+    command: BOUND
+    affinity_key: "session"
+  }
+}
+method: {
+  name: "/google.spanner.v1.Spanner/Rollback"
+  affinity: {
+    command: BOUND
+    affinity_key: "session"
+  }
+}
+method: {
+  name: "/google.spanner.v1.Spanner/PartitionQuery"
+  affinity: {
+    command: BOUND
+    affinity_key: "session"
+  }
+}
+method: {
+  name: "/google.spanner.v1.Spanner/PartitionRead"
+  affinity: {
+    command: BOUND
+    affinity_key: "session"
+  }
+}

spanner/google/cloud/spanner_v1/gapic/spanner_client.py

@@ -32,8 +32,15 @@
 from google.cloud.spanner_v1.proto import transaction_pb2
 from google.protobuf import struct_pb2
 
+try:
+    import grpc_gcp
+    HAS_GRPC_GCP = True
+except ImportError:
+    HAS_GRPC_GCP = False
+
 _GAPIC_LIBRARY_VERSION = pkg_resources.get_distribution(
     'google-cloud-spanner', ).version
+_SPANNER_GRPC_CONFIG = 'spanner.grpc.config'
 
 
 class SpannerClient(object):
@@ -113,10 +120,20 @@ def __init__(self,
 
         # Create the channel.
         if channel is None:
+            options = None
+
+            if HAS_GRPC_GCP:
+                # Initialize grpc gcp config for spanner api.
+                grpc_gcp_config = grpc_gcp.api_config_from_text_pb(
+                    pkg_resources.resource_string(__name__,
+                                                  _SPANNER_GRPC_CONFIG))
+                options = [(grpc_gcp.API_CONFIG_CHANNEL_ARG, grpc_gcp_config)]
+
             channel = google.api_core.grpc_helpers.create_channel(
                 self.SERVICE_ADDRESS,
                 credentials=credentials,
                 scopes=self._DEFAULT_SCOPES,
+                options=options,
             )
 
         # Create the gRPC stubs.