Merge pull request #1931 from tseaver/bigtable-v2-happybase

Convert `gcloud.bigtable.happybase` to V2 patterns

Author: tseaver

gcloud/bigtable/happybase/__init__.py

@@ -84,9 +84,9 @@
   * ``protocol``
 * In order to make
   :class:`Connection <gcloud.bigtable.happybase.connection.Connection>`
-  compatible with Cloud Bigtable, we add a ``cluster`` keyword argument to
+  compatible with Cloud Bigtable, we add a ``instance`` keyword argument to
   allow users to pass in their own
-  :class:`Cluster <gcloud.bigtable.cluster.Cluster>` (which they can
+  :class:`Instance <gcloud.bigtable.instance.Instance>` (which they can
   construct beforehand).
 
   For example:
@@ -95,11 +95,11 @@
 
       from gcloud.bigtable.client import Client
       client = Client(project=PROJECT_ID, admin=True)
-      cluster = client.cluster(zone, cluster_id)
-      cluster.reload()
+      instance = client.instance(instance_id, location_id)
+      instance.reload()
 
       from gcloud.bigtable.happybase import Connection
-      connection = Connection(cluster=cluster)
+      connection = Connection(instance=instance)
 
 * Any uses of the ``wal`` (Write Ahead Log) argument will result in a
   warning as well. This includes uses in:

gcloud/bigtable/happybase/connection.py

@@ -54,57 +54,57 @@
                        'of enabled / disabled tables.')
 
 
-def _get_cluster(timeout=None):
-    """Gets cluster for the default project.
+def _get_instance(timeout=None):
+    """Gets instance for the default project.
 
     Creates a client with the inferred credentials and project ID from
     the local environment. Then uses
-    :meth:`.bigtable.client.Client.list_clusters` to
-    get the unique cluster owned by the project.
+    :meth:`.bigtable.client.Client.list_instances` to
+    get the unique instance owned by the project.
 
-    If the request fails for any reason, or if there isn't exactly one cluster
+    If the request fails for any reason, or if there isn't exactly one instance
     owned by the project, then this function will fail.
 
     :type timeout: int
     :param timeout: (Optional) The socket timeout in milliseconds.
 
-    :rtype: :class:`gcloud.bigtable.cluster.Cluster`
-    :returns: The unique cluster owned by the project inferred from
+    :rtype: :class:`gcloud.bigtable.instance.Instance`
+    :returns: The unique instance owned by the project inferred from
               the environment.
     :raises: :class:`ValueError <exceptions.ValueError>` if there is a failed
-             zone or any number of clusters other than one.
+             location or any number of instances other than one.
     """
     client_kwargs = {'admin': True}
     if timeout is not None:
         client_kwargs['timeout_seconds'] = timeout / 1000.0
     client = Client(**client_kwargs)
     try:
         client.start()
-        clusters, failed_zones = client.list_clusters()
+        instances, failed_locations = client.list_instances()
     finally:
         client.stop()
 
-    if len(failed_zones) != 0:
-        raise ValueError('Determining cluster via ListClusters encountered '
-                         'failed zones.')
-    if len(clusters) == 0:
-        raise ValueError('This client doesn\'t have access to any clusters.')
-    if len(clusters) > 1:
-        raise ValueError('This client has access to more than one cluster. '
-                         'Please directly pass the cluster you\'d '
+    if len(failed_locations) != 0:
+        raise ValueError('Determining instance via ListInstances encountered '
+                         'failed locations.')
+    if len(instances) == 0:
+        raise ValueError('This client doesn\'t have access to any instances.')
+    if len(instances) > 1:
+        raise ValueError('This client has access to more than one instance. '
+                         'Please directly pass the instance you\'d '
                          'like to use.')
-    return clusters[0]
+    return instances[0]
 
 
 class Connection(object):
     """Connection to Cloud Bigtable backend.
 
     .. note::
 
-        If you pass a ``cluster``, it will be :meth:`.Cluster.copy`-ed before
+        If you pass a ``instance``, it will be :meth:`.Instance.copy`-ed before
         being stored on the new connection. This also copies the
         :class:`Client <gcloud.bigtable.client.Client>` that created the
-        :class:`Cluster <gcloud.bigtable.cluster.Cluster>` instance and the
+        :class:`Instance <gcloud.bigtable.instance.Instance>` instance and the
         :class:`Credentials <oauth2client.client.Credentials>` stored on the
         client.
 
@@ -127,27 +127,27 @@ class Connection(object):
     :param table_prefix_separator: (Optional) Separator used with
                                    ``table_prefix``. Defaults to ``_``.
 
-    :type cluster: :class:`Cluster <gcloud.bigtable.cluster.Cluster>`
-    :param cluster: (Optional) A Cloud Bigtable cluster. The instance also
+    :type instance: :class:`Instance <gcloud.bigtable.instance.Instance>`
+    :param instance: (Optional) A Cloud Bigtable instance. The instance also
                     owns a client for making gRPC requests to the Cloud
                     Bigtable API. If not passed in, defaults to creating client
                     with ``admin=True`` and using the ``timeout`` here for the
                     ``timeout_seconds`` argument to the
                     :class:`Client <gcloud.bigtable.client.Client>`
                     constructor. The credentials for the client
                     will be the implicit ones loaded from the environment.
-                    Then that client is used to retrieve all the clusters
+                    Then that client is used to retrieve all the instances
                     owned by the client's project.
 
     :type kwargs: dict
     :param kwargs: Remaining keyword arguments. Provided for HappyBase
                    compatibility.
     """
 
-    _cluster = None
+    _instance = None
 
     def __init__(self, timeout=None, autoconnect=True, table_prefix=None,
-                 table_prefix_separator='_', cluster=None, **kwargs):
+                 table_prefix_separator='_', instance=None, **kwargs):
         self._handle_legacy_args(kwargs)
         if table_prefix is not None:
             if not isinstance(table_prefix, six.string_types):
@@ -162,13 +162,13 @@ def __init__(self, timeout=None, autoconnect=True, table_prefix=None,
         self.table_prefix = table_prefix
         self.table_prefix_separator = table_prefix_separator
 
-        if cluster is None:
-            self._cluster = _get_cluster(timeout=timeout)
+        if instance is None:
+            self._instance = _get_instance(timeout=timeout)
         else:
             if timeout is not None:
                 raise ValueError('Timeout cannot be used when an existing '
-                                 'cluster is passed')
-            self._cluster = cluster.copy()
+                                 'instance is passed')
+            self._instance = instance.copy()
 
         if autoconnect:
             self.open()
@@ -203,23 +203,23 @@ def open(self):
 
         This method opens the underlying HTTP/2 gRPC connection using a
         :class:`Client <gcloud.bigtable.client.Client>` bound to the
-        :class:`Cluster <gcloud.bigtable.cluster.Cluster>` owned by
+        :class:`Instance <gcloud.bigtable.instance.Instance>` owned by
         this connection.
         """
-        self._cluster._client.start()
+        self._instance._client.start()
 
     def close(self):
         """Close the underlying transport to Cloud Bigtable.
 
         This method closes the underlying HTTP/2 gRPC connection using a
         :class:`Client <gcloud.bigtable.client.Client>` bound to the
-        :class:`Cluster <gcloud.bigtable.cluster.Cluster>` owned by
+        :class:`Instance <gcloud.bigtable.instance.Instance>` owned by
         this connection.
         """
-        self._cluster._client.stop()
+        self._instance._client.stop()
 
     def __del__(self):
-        if self._cluster is not None:
+        if self._instance is not None:
             self.close()
 
     def _table_name(self, name):
@@ -258,7 +258,7 @@ def tables(self):
 
         .. note::
 
-            This lists every table in the cluster owned by this connection,
+            This lists every table in the instance owned by this connection,
             **not** every table that a given user may have access to.
 
         .. note::
@@ -269,7 +269,7 @@ def tables(self):
         :rtype: list
         :returns: List of string table names.
         """
-        low_level_table_instances = self._cluster.list_tables()
+        low_level_table_instances = self._instance.list_tables()
         table_names = [table_instance.table_id
                        for table_instance in low_level_table_instances]
 
@@ -345,7 +345,7 @@ def create_table(self, name, families):
 
         # Create table instance and then make API calls.
         name = self._table_name(name)
-        low_level_table = _LowLevelTable(name, self._cluster)
+        low_level_table = _LowLevelTable(name, self._instance)
         try:
             low_level_table.create()
         except face.NetworkError as network_err:
@@ -376,7 +376,7 @@ def delete_table(self, name, disable=False):
             _WARN(_DISABLE_DELETE_MSG)
 
         name = self._table_name(name)
-        _LowLevelTable(name, self._cluster).delete()
+        _LowLevelTable(name, self._instance).delete()
 
     def enable_table(self, name):
         """Enable the specified table.

gcloud/bigtable/happybase/pool.py

@@ -21,7 +21,7 @@
 import six
 
 from gcloud.bigtable.happybase.connection import Connection
-from gcloud.bigtable.happybase.connection import _get_cluster
+from gcloud.bigtable.happybase.connection import _get_instance
 
 
 _MIN_POOL_SIZE = 1
@@ -45,7 +45,7 @@ class ConnectionPool(object):
         :class:`Connection <.happybase.connection.Connection>` constructor
         **except** for ``autoconnect``. This is because the ``open`` /
         ``closed`` status of a connection is managed by the pool. In addition,
-        if ``cluster`` is not passed, the default / inferred cluster is
+        if ``instance`` is not passed, the default / inferred instance is
         determined by the pool and then passed to each
         :class:`Connection <.happybase.connection.Connection>` that is created.
 
@@ -75,8 +75,8 @@ def __init__(self, size, **kwargs):
 
         connection_kwargs = kwargs
         connection_kwargs['autoconnect'] = False
-        if 'cluster' not in connection_kwargs:
-            connection_kwargs['cluster'] = _get_cluster(
+        if 'instance' not in connection_kwargs:
+            connection_kwargs['instance'] = _get_instance(
                 timeout=kwargs.get('timeout'))
 
         for _ in six.moves.range(size):

gcloud/bigtable/happybase/table.py

@@ -109,13 +109,13 @@ class Table(object):
 
     def __init__(self, name, connection):
         self.name = name
-        # This remains as legacy for HappyBase, but only the cluster
+        # This remains as legacy for HappyBase, but only the instance
         # from the connection is needed.
         self.connection = connection
         self._low_level_table = None
         if self.connection is not None:
             self._low_level_table = _LowLevelTable(self.name,
-                                                   self.connection._cluster)
+                                                   self.connection._instance)
 
     def __repr__(self):
         return '<table.Table name=%r>' % (self.name,)
@@ -378,42 +378,8 @@ def scan(self, row_start=None, row_stop=None, row_prefix=None,
                  :class:`TypeError <exceptions.TypeError>` if a string
                  ``filter`` is used.
         """
-        filter_ = kwargs.pop('filter', None)
-        legacy_args = []
-        for kw_name in ('batch_size', 'scan_batching', 'sorted_columns'):
-            if kw_name in kwargs:
-                legacy_args.append(kw_name)
-                kwargs.pop(kw_name)
-        if legacy_args:
-            legacy_args = ', '.join(legacy_args)
-            message = ('The HappyBase legacy arguments %s were used. These '
-                       'arguments are unused by gcloud.' % (legacy_args,))
-            _WARN(message)
-        if kwargs:
-            raise TypeError('Received unexpected arguments', kwargs.keys())
-
-        if limit is not None and limit < 1:
-            raise ValueError('limit must be positive')
-        if row_prefix is not None:
-            if row_start is not None or row_stop is not None:
-                raise ValueError('row_prefix cannot be combined with '
-                                 'row_start or row_stop')
-            row_start = row_prefix
-            row_stop = _string_successor(row_prefix)
-
-        filters = []
-        if isinstance(filter_, six.string_types):
-            raise TypeError('Specifying filters as a string is not supported '
-                            'by Cloud Bigtable. Use a '
-                            'gcloud.bigtable.row.RowFilter instead.')
-        elif filter_ is not None:
-            filters.append(filter_)
-
-        if columns is not None:
-            filters.append(_columns_filter_helper(columns))
-        # versions == 1 since we only want the latest.
-        filter_chain = _filter_chain_helper(versions=1, timestamp=timestamp,
-                                            filters=filters)
+        row_start, row_stop, filter_chain = _scan_filter_helper(
+            row_start, row_stop, row_prefix, columns, timestamp, limit, kwargs)
 
         partial_rows_data = self._low_level_table.read_rows(
             start_key=row_start, end_key=row_stop,
@@ -424,11 +390,12 @@ def scan(self, row_start=None, row_stop=None, row_prefix=None,
         while True:
             try:
                 partial_rows_data.consume_next()
-                row_key, curr_row_data = rows_dict.popitem()
-                # NOTE: We expect len(rows_dict) == 0, but don't check it.
-                curr_row_dict = _partial_row_to_dict(
-                    curr_row_data, include_timestamp=include_timestamp)
-                yield (row_key, curr_row_dict)
+                for row_key in sorted(rows_dict):
+                    curr_row_data = rows_dict.pop(row_key)
+                    # NOTE: We expect len(rows_dict) == 0, but don't check it.
+                    curr_row_dict = _partial_row_to_dict(
+                        curr_row_data, include_timestamp=include_timestamp)
+                    yield (row_key, curr_row_dict)
             except StopIteration:
                 break
 
@@ -911,6 +878,49 @@ def _filter_chain_helper(column=None, versions=None, timestamp=None,
         return RowFilterChain(filters=filters)
 
 
+def _scan_filter_helper(row_start, row_stop, row_prefix, columns,
+                        timestamp, limit, kwargs):
+    """Helper for :meth:`scan`:  build up a filter chain."""
+    filter_ = kwargs.pop('filter', None)
+    legacy_args = []
+    for kw_name in ('batch_size', 'scan_batching', 'sorted_columns'):
+        if kw_name in kwargs:
+            legacy_args.append(kw_name)
+            kwargs.pop(kw_name)
+    if legacy_args:
+        legacy_args = ', '.join(legacy_args)
+        message = ('The HappyBase legacy arguments %s were used. These '
+                   'arguments are unused by gcloud.' % (legacy_args,))
+        _WARN(message)
+    if kwargs:
+        raise TypeError('Received unexpected arguments', kwargs.keys())
+
+    if limit is not None and limit < 1:
+        raise ValueError('limit must be positive')
+    if row_prefix is not None:
+        if row_start is not None or row_stop is not None:
+            raise ValueError('row_prefix cannot be combined with '
+                             'row_start or row_stop')
+        row_start = row_prefix
+        row_stop = _string_successor(row_prefix)
+
+    filters = []
+    if isinstance(filter_, six.string_types):
+        raise TypeError('Specifying filters as a string is not supported '
+                        'by Cloud Bigtable. Use a '
+                        'gcloud.bigtable.row.RowFilter instead.')
+    elif filter_ is not None:
+        filters.append(filter_)
+
+    if columns is not None:
+        filters.append(_columns_filter_helper(columns))
+
+    # versions == 1 since we only want the latest.
+    filter_ = _filter_chain_helper(versions=1, timestamp=timestamp,
+                                   filters=filters)
+    return row_start, row_stop, filter_
+
+
 def _columns_filter_helper(columns):
     """Creates a union filter for a list of columns.