allow enabling queued/async iterator with PYGLOSSARY_ASYNC_ITER_SIZE env

pyglossary/glossary_v2.py

@@ -34,6 +34,8 @@
 from typing import TYPE_CHECKING, cast
 from uuid import uuid1
 
+from pyglossary.queued_iter import QueuedIterator
+
 from . import core
 from .core import (
 	cacheDir,
@@ -440,6 +442,11 @@ def _readersEntryGen(self) -> Iterator[EntryType]:
 
 			iterator = self._applyEntryFiltersGen(iterator)
 
+			# turn iterator into background-queued, like buffered channel in Go
+			queueSize = os.getenv("PYGLOSSARY_ASYNC_ITER_SIZE")
+			if queueSize:
+				iterator = QueuedIterator(iterator, int(queueSize))
+
 			try:
 				yield from iterator
 			finally:

pyglossary/queued_iter.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+import queue
+import threading
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+	from collections.abc import Iterator
+
+
+class QueuedIterator:
+	def __init__(
+		self,
+		iterator: Iterator,
+		max_size: int,
+	) -> None:
+		self.iterator = iterator
+		self.queue = queue.Queue(max_size)
+		self.thread = threading.Thread(target=self._background_job)
+		self.thread.start()
+
+	def _background_job(self) -> None:
+		for item in self.iterator:
+			self.queue.put(item)
+		self.queue.put(StopIteration)
+
+	def __iter__(self) -> Iterator:
+		return self
+
+	def __next__(self) -> Any:
+		item = self.queue.get()
+		if item is StopIteration:
+			raise StopIteration
+		return item