black

streamz_ext/clients.py

@@ -13,7 +13,7 @@ def result_maybe(future_maybe):
         return future_maybe.result()
     except AttributeError:
         if isinstance(future_maybe, Sequence) and not isinstance(
-                future_maybe, str
+            future_maybe, str
         ):
             aa = []
             for a in future_maybe:

streamz_ext/core.py

@@ -288,5 +288,3 @@ def destroy_pipeline(source_node: Stream):
         # some source nodes are tuples and some are bad wekrefs
         except (AttributeError, KeyError) as e:
             pass
-
-

test/test_parallel.py

@@ -324,17 +324,16 @@ def test_combined_latest(backend):
     def delay(x):
         time.sleep(.5)
         return x
+
     source = Stream(asynchronous=True)
     source2 = Stream(asynchronous=True)
-    futures = source.scatter(backend=backend).map(delay).combine_latest(
-        source2.scatter(backend=backend), emit_on=1)
-    futures_L = futures.sink_to_list()
-    L = (
-        futures
-        .buffer(10)
-        .gather()
-        .sink_to_list()
+    futures = (
+        source.scatter(backend=backend)
+        .map(delay)
+        .combine_latest(source2.scatter(backend=backend), emit_on=1)
     )
+    futures_L = futures.sink_to_list()
+    L = futures.buffer(10).gather().sink_to_list()
 
     for i in range(5):
         yield source.emit(i)

test/test_parallel_dask.py

@@ -265,9 +265,9 @@ def test_double_scatter(c, s, a, b):
     r = sm.buffer(10).gather()
     L = r.sink_to_list()
 
-    print('hi')
+    print("hi")
     for i in range(5):
-        print('hi')
+        print("hi")
         yield source1.emit(i)
         yield source2.emit(i)