Merge pull request #402 from nodestream-proj/concurrent-transformer-bugfix

fix: ConcurrentTransformer missing arg

Author: zprobst

nodestream/pipeline/transformers/transformer.py

@@ -1,5 +1,5 @@
 import asyncio
-from abc import abstractmethod
+from abc import ABC, abstractmethod
 from concurrent.futures import ThreadPoolExecutor
 from logging import getLogger
 from typing import Any, AsyncGenerator, Dict, Optional
@@ -34,7 +34,7 @@ async def transform_record(self, record: Any) -> Any:
         raise NotImplementedError
 
 
-class ConcurrentTransformer(Transformer):
+class ConcurrentTransformer(Transformer, ABC):
     """A `ConcurrentTransformer` can process multiple records concurrently.
 
     `ConcurrentTransformer`s are useful when the transformation work is IO
@@ -60,7 +60,7 @@ def __init__(
         self.pending_tasks = []
 
     def on_worker_thread_start(self):
-        pass
+        pass  # do nothing
 
     async def drain_completed_tasks(self):
         tasks_drained = 0
@@ -107,7 +107,7 @@ def do_work_on_record(self, record: Any) -> Any:
     async def yield_processor(self):
         await asyncio.sleep(0)
 
-    async def emit_outstanding_records(self):
+    async def emit_outstanding_records(self, context: StepContext):
         while self.pending_tasks:
             async for result in self.drain_completed_tasks():
                 yield result