diff --git a/CHANGELOG.md b/CHANGELOG.md index 4a559f0..9887c6e 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/_python_utils_tests/test_generators.py b/_python_utils_tests/test_generators.py index 19a9bcc..0eaa1b0 100644 --- a/_python_utils_tests/test_generators.py +++ b/_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 = [] diff --git a/python_utils/generators.py b/python_utils/generators.py index 093cbf3..bc83080 100644 --- a/python_utils/generators.py +++ b/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)