Single Client Reuse

nodestream/databases/debounced_ingest_strategy.py

@@ -74,3 +74,7 @@ async def flush(self):
         for hook in self.hooks_saved_for_after_ingest:
             await self.executor.execute_hook(hook)
         self.hooks_saved_for_after_ingest.clear()
+
+    async def finish(self):
+        """Close connector by calling finish method from Step"""
+        await self.executor.finish()

nodestream/databases/ingest_strategy.py

@@ -64,3 +64,7 @@ async def perform_ttl_operation(self, config: "TimeToLiveConfiguration"):
     async def flush(self):
         """Flush any pending operations to the database."""
         pass
+
+    async def finish(self):
+        """Close connector by calling finish method from Step"""
+        self.connector.finish()

nodestream/databases/query_executor_with_statistics.py

@@ -43,3 +43,6 @@ async def perform_ttl_op(self, config: TimeToLiveConfiguration):
     async def execute_hook(self, hook: IngestionHook):
         await self.inner.execute_hook(hook)
         get_context().increment_stat(HOOK_STAT)
+
+    async def finish(self):
+        await self.inner.finish()

nodestream/databases/writer.py

@@ -59,3 +59,8 @@ async def write_record(self, ingestible):
         self.pending_records += 1
         if self.pending_records >= self.batch_size:
             await self.flush()
+
+    async def finish(self):
+        """Close connector by calling finish method from Step"""
+        await self.flush()
+        await self.ingest_strategy.finish()