feat: add BatchProcessor utility class

Author: LifeJiggy

src/gradient/_utils/__init__.py

@@ -34,6 +34,7 @@
     validate_client_instance as validate_client_instance,
     ResponseCache as ResponseCache,
     RateLimiter as RateLimiter,
+    BatchProcessor as BatchProcessor,
 )
 from ._compat import (
     get_args as get_args,

src/gradient/_utils/_utils.py

@@ -544,6 +544,74 @@ def wait_time(self, tokens: int = 1) -> float:
         return needed / self.refill_rate
 
 
+# Batch Processing Classes
+class BatchProcessor:
+    """Utility for batching multiple requests with timeout and size limits."""
+
+    def __init__(self, batch_size: int = 10, timeout_seconds: float = 5.0) -> None:
+        """Initialize batch processor.
+
+        Args:
+            batch_size: Maximum items per batch
+            timeout_seconds: Maximum time to wait before processing batch
+        """
+        self.batch_size: int = batch_size
+        self.timeout_seconds: float = timeout_seconds
+        self._batch: list[Any] = []
+        self._last_add_time: float = self._now()
+        self._callback: Callable[[list[Any]], Any] | None = None
+
+    def _now(self) -> float:
+        """Get current time in seconds."""
+        import time
+        return time.time()
+
+    def add(self, item: Any) -> None:
+        """Add item to current batch."""
+        self._batch.append(item)
+        self._last_add_time = self._now()
+
+        # Auto-process if batch is full
+        if len(self._batch) >= self.batch_size:
+            self._process_batch()
+
+    def set_callback(self, callback: Callable[[list[Any]], Any]) -> None:
+        """Set callback function to process batches."""
+        self._callback = callback
+
+    def _process_batch(self) -> Any | None:
+        """Process current batch if not empty."""
+        if not self._batch or not self._callback:
+            return None
+
+        batch = self._batch.copy()
+        self._batch.clear()
+        return self._callback(batch)
+
+    def force_process(self) -> Any | None:
+        """Force process current batch regardless of size or timeout."""
+        return self._process_batch()
+
+    def check_timeout(self) -> Any | None:
+        """Check if batch has timed out and process if needed."""
+        if not self._batch:
+            return None
+
+        elapsed = self._now() - self._last_add_time
+        if elapsed >= self.timeout_seconds:
+            return self._process_batch()
+
+        return None
+
+    def size(self) -> int:
+        """Get current batch size."""
+        return len(self._batch)
+
+    def is_empty(self) -> bool:
+        """Check if batch is empty."""
+        return len(self._batch) == 0
+
+
 # API Key Validation Functions
 def validate_api_key(api_key: str | None) -> bool:
     """Validate an API key format.

tests/test_batch_processor.py

@@ -0,0 +1,91 @@
+"""Tests for batch processing functionality."""
+
+import time
+import pytest
+from gradient._utils import BatchProcessor
+
+
+class TestBatchProcessor:
+    """Test batch processing functionality."""
+
+    def test_batch_processor_basic(self):
+        """Test basic batch processing."""
+        processor = BatchProcessor(batch_size=3)
+        processed_batches = []
+
+        def process_batch(batch):
+            processed_batches.append(batch)
+            return f"processed {len(batch)} items"
+
+        processor.set_callback(process_batch)
+
+        # Add items
+        processor.add("item1")
+        processor.add("item2")
+        assert processor.size() == 2
+
+        # Add third item to trigger auto-processing
+        processor.add("item3")
+        assert processor.size() == 0  # Should be cleared after processing
+        assert len(processed_batches) == 1
+        assert processed_batches[0] == ["item1", "item2", "item3"]
+
+    def test_batch_processor_timeout(self):
+        """Test batch processing with timeout."""
+        processor = BatchProcessor(batch_size=10, timeout_seconds=0.1)
+        processed_batches = []
+
+        def process_batch(batch):
+            processed_batches.append(batch)
+
+        processor.set_callback(process_batch)
+
+        # Add item and wait for timeout
+        processor.add("item1")
+        time.sleep(0.2)
+
+        # Check timeout should process batch
+        processor.check_timeout()
+        assert len(processed_batches) == 1
+        assert processed_batches[0] == ["item1"]
+
+    def test_batch_processor_force_process(self):
+        """Test force processing of batch."""
+        processor = BatchProcessor(batch_size=10)
+        processed_batches = []
+
+        def process_batch(batch):
+            processed_batches.append(batch)
+
+        processor.set_callback(process_batch)
+
+        # Add items without reaching batch size
+        processor.add("item1")
+        processor.add("item2")
+        assert processor.size() == 2
+
+        # Force process
+        processor.force_process()
+        assert processor.size() == 0
+        assert len(processed_batches) == 1
+        assert processed_batches[0] == ["item1", "item2"]
+
+    def test_batch_processor_multiple_batches(self):
+        """Test processing multiple batches."""
+        processor = BatchProcessor(batch_size=2)
+        processed_batches = []
+
+        def process_batch(batch):
+            processed_batches.append(batch)
+
+        processor.set_callback(process_batch)
+
+        # Add items to create multiple batches
+        processor.add("item1")
+        processor.add("item2")  # Triggers first batch
+        processor.add("item3")
+        processor.add("item4")  # Triggers second batch
+
+        assert len(processed_batches) == 2
+        assert processed_batches[0] == ["item1", "item2"]
+        assert processed_batches[1] == ["item3", "item4"]