Merge upstream 05 03 26

faust/__init__.py

@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 """Python Stream processing."""
+
 # :copyright: (c) 2017-2020, Robinhood Markets, Inc.
 #             All rights reserved.
 # :license:   BSD (3 Clause), see LICENSE for more details.

faust/stores/__init__.py

@@ -12,6 +12,7 @@
     memory="faust.stores.memory:Store",
     rocksdb="faust.stores.rocksdb:Store",
     aerospike="faust.stores.aerospike:AeroSpikeStore",
+    bigtable="faust.stores.bigtable:BigTableStore",
 )
 STORES.include_setuptools_namespace("faust.stores")
 by_name = STORES.by_name

faust/stores/aerospike.py

@@ -98,7 +98,7 @@ def _get(self, key: bytes) -> Optional[bytes]:
         key = (self.namespace, self.table_name, key)
         fun = self.client.get
         try:
-            (key, meta, bins) = self.aerospike_fun_call_with_retry(fun=fun, key=key)
+            key, meta, bins = self.aerospike_fun_call_with_retry(fun=fun, key=key)
             if bins:
                 return bins[self.BIN_KEY]
             return None
@@ -173,7 +173,7 @@ def _itervalues(self) -> Iterator[bytes]:
                 fun=fun, namespace=self.namespace, set=self.table_name
             )
             for result in scan.results():
-                (key, meta, bins) = result
+                key, meta, bins = result
                 if bins:
                     yield bins[self.BIN_KEY]
                 else:
@@ -193,8 +193,8 @@ def _iteritems(self) -> Iterator[Tuple[bytes, bytes]]:
                 fun=fun, namespace=self.namespace, set=self.table_name
             )
             for result in scan.results():
-                (key_data, meta, bins) = result
-                (ns, set, policy, key) = key_data
+                key_data, meta, bins = result
+                ns, set, policy, key = key_data
 
                 if bins:
                     bins = bins[self.BIN_KEY]
@@ -214,7 +214,7 @@ def _contains(self, key: bytes) -> bool:
         try:
             if self.app.conf.store_check_exists:
                 key = (self.namespace, self.table_name, key)
-                (key, meta) = self.aerospike_fun_call_with_retry(
+                key, meta = self.aerospike_fun_call_with_retry(
                     fun=self.client.exists, key=key
                 )
                 if meta: