fix: deprecate nonpositive batch sizes before 5.0

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## Unreleased
+
+- Deprecate batch sizes below 1 in `batcher` and in `abatcher` without an
+  interval. A `DeprecationWarning` is issued before consuming the source;
+  existing results and errors are preserved in 4.x. These sizes will raise
+  `ValueError` in 5.0. With an interval, `abatcher` continues to accept `None`
+  or a size below 1 to mean no size limit.
+
 ## 4.1.4 - 2026-10-21
 
 - Build from the source distribution with `uv_build` 0.13 and later. The

_python_utils_tests/test_generators.py

@@ -4,6 +4,7 @@
 import datetime
 import gc
 import sys
+import warnings
 from types import SimpleNamespace
 
 import pytest
@@ -99,7 +100,7 @@ async def generator() -> types.AsyncIterator[int]:
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize('batch_size', [None, 3])
+@pytest.mark.parametrize('batch_size', [None, 0, -1, 3])
 @pytest.mark.parametrize('interval', [10, datetime.timedelta(seconds=10)])
 @pytest.mark.parametrize('arrival', [8.0, 10.0])
 async def test_abatcher_partial_batch_keeps_deadline(
@@ -364,6 +365,111 @@ async def test_abatcher_with_future_returning_iterator() -> None:
     assert batches == [[0, 1], [2, 3], [4]]
 
 
+@pytest.mark.parametrize('batch_size', [0, -1])
+@pytest.mark.parametrize('count', [0, 3])
+def test_batcher_warns_for_nonpositive_size(
+    batch_size: int, count: int
+) -> None:
+    """Warn once while keeping the existing result and consumption order."""
+    consumed: list[int] = []
+
+    def source() -> types.Generator[int, None, None]:
+        for item in range(count):
+            consumed.append(item)
+            yield item
+
+    batcher = python_utils.batcher(source(), batch_size)
+    assert consumed == []
+    with pytest.warns(DeprecationWarning, match='ValueError.*5.0') as caught:
+        result = list(batcher)
+
+    assert result == ([list(range(count))] if count else [])
+    assert consumed == list(range(count))
+    assert len(caught) == 1
+    assert caught[0].filename == __file__
+
+
+@pytest.mark.parametrize('batch_size', [0, -1])
+def test_batcher_warning_as_error_does_not_consume(batch_size: int) -> None:
+    """Let callers opt into failure before taking the first source item."""
+    source = iter(range(3))
+    batcher = python_utils.batcher(source, batch_size)
+    with warnings.catch_warnings():
+        warnings.simplefilter('error', DeprecationWarning)
+        with pytest.raises(DeprecationWarning, match='batch_size'):
+            next(batcher)
+    assert list(source) == [0, 1, 2]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('batch_size', [0, -1])
+@pytest.mark.parametrize('count', [0, 3])
+@pytest.mark.parametrize('interval', [None, 0])
+async def test_abatcher_warns_without_interval(
+    batch_size: int, count: int, interval: int | None
+) -> None:
+    """Keep the old zero-size error and negative-size result in 4.x."""
+    consumed: list[int] = []
+
+    async def source() -> types.AsyncGenerator[int, None]:
+        for item in range(count):
+            consumed.append(item)
+            yield item
+
+    batcher = python_utils.abatcher(source(), batch_size, interval)
+    assert consumed == []
+    with pytest.warns(DeprecationWarning, match='ValueError.*5.0') as caught:
+        if batch_size == 0:
+            with pytest.raises(AssertionError, match='Must specify either'):
+                await batcher.__anext__()
+            assert consumed == []
+        else:
+            result = [batch async for batch in batcher]
+            assert result == ([list(range(count))] if count else [])
+            assert consumed == list(range(count))
+
+    assert len(caught) == 1
+    assert caught[0].filename == __file__
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('batch_size', [0, -1])
+async def test_abatcher_warning_as_error_does_not_consume(
+    batch_size: int,
+) -> None:
+    """Warn before scheduling even the first asynchronous source item."""
+    source = FutureIterator(3)
+    batcher = python_utils.abatcher(source, batch_size)
+    with warnings.catch_warnings():
+        warnings.simplefilter('error', DeprecationWarning)
+        with pytest.raises(DeprecationWarning, match='batch_size'):
+            await batcher.__anext__()
+    assert source.current == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('batch_size', [None, 0, -1])
+@pytest.mark.parametrize('count', [0, 3])
+async def test_abatcher_interval_only_sizes(
+    batch_size: int | None, count: int
+) -> None:
+    """None and nonpositive sizes impose no limit and issue no warning."""
+    source = FutureIterator(count)
+    result = [
+        batch async for batch in python_utils.abatcher(source, batch_size, 10)
+    ]
+    assert result == ([list(range(count))] if count else [])
+    assert source.current == count
+
+
+def test_batcher_positive_size_is_lazy() -> None:
+    """Positive sizes issue no warning and consume only the first batch."""
+    source = iter(range(3))
+    batcher = python_utils.batcher(source, 1)
+    assert next(batcher) == [0]
+    assert list(source) == [1, 2]
+
+
 def test_batcher() -> None:
     """Split an iterable into fixed-size batches."""
     batch = []

python_utils/generators.py

@@ -15,6 +15,7 @@
 import collections.abc
 import time
 import typing
+import warnings
 
 import python_utils
 from python_utils import _aliases
@@ -23,6 +24,22 @@
 _T = typing.TypeVar('_T')
 
 
+def _warn_nonpositive_batch_size(
+    batch_size: int | None, interval: _aliases.delta_type | None = None
+) -> int | None:
+    """Warn for a deprecated size, or use interval-only batching."""
+    if batch_size is not None and batch_size < 1:
+        if interval:
+            return None
+        warnings.warn(
+            'batch_size must be greater than zero; sizes below 1 are '
+            'deprecated and will raise ValueError in python-utils 5.0',
+            DeprecationWarning,
+            stacklevel=3,
+        )
+    return batch_size
+
+
 async def _cancel(pending: set[asyncio.Future[_T]]) -> None:
     """Cancel the items that are still on their way and wait for them.
 
@@ -78,14 +95,22 @@ async def abatcher(
     Args:
         generator: The async generator or iterator to batch.
         batch_size (typing.Optional[int], optional): The number of items per
-            batch. Defaults to None.
+            batch. Defaults to None. With an interval, None or a value below
+            1 means no size limit. Without an interval, sizes below 1 are
+            deprecated and will raise ValueError in version 5.0.
         interval (typing.Optional[_aliases.delta_type], optional): The time
             interval to wait before yielding a batch. Defaults to None.
 
     Yields:
         collections.abc.AsyncGenerator[list[_T], None]: A generator that yields
         batches of items.
+
+    Warns:
+        DeprecationWarning: If batch_size is below 1 without an interval.
+            The existing behavior is preserved until version 5.0.
     """
+    batch_size = _warn_nonpositive_batch_size(batch_size, interval)
+
     batch: list[_T] = []
 
     assert batch_size or interval, 'Must specify either batch_size or interval'
@@ -170,12 +195,20 @@ def batcher(
     Args:
         iterable (collections.abc.Iterable[_T]): The iterable to batch.
         batch_size (int, optional): The number of items per batch. Defaults
-            to 10.
+            to 10. Sizes below 1 are deprecated and will raise ValueError in
+            version 5.0. Until then, they yield all items in a single batch,
+            or no batches for an empty iterable.
 
     Yields:
         collections.abc.Generator[list[_T], None, None]: A generator that
             yields batches of items.
+
+    Warns:
+        DeprecationWarning: If batch_size is below 1. The warning is issued
+            when iteration begins, before consuming any source items.
     """
+    _warn_nonpositive_batch_size(batch_size)
+
     batch: list[_T] = []
     for item in iterable:
         batch.append(item)