From 37f74e49758f7be6928ba782c9e875f74819efee Mon Sep 17 00:00:00 2001 From: tongwentao Date: Thu, 22 Oct 2026 16:12:01 +0800 Subject: [PATCH 1/3] fix: reject nonpositive batch sizes before consuming items --- CHANGELOG.md | 7 +++++ _python_utils_tests/test_generators.py | 43 ++++++++++++++++++++++++++ python_utils/generators.py | 20 ++++++++++-- 3 files changed, 68 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4a559f0..e474c71 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## Unreleased + +- Reject zero and negative batch sizes with `ValueError` in `batcher` and + `abatcher` before consuming the source. These sizes never reached the batch + limit and could buffer an unbounded source indefinitely. `abatcher` still + accepts `None` for interval-only batching. + ## 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..f0ca8cb 100644 --- a/_python_utils_tests/test_generators.py +++ b/_python_utils_tests/test_generators.py @@ -364,6 +364,49 @@ 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_rejects_nonpositive_size(batch_size: int, count: int) -> None: + """Reject invalid sizes without consuming even a finite source.""" + consumed: list[int] = [] + + def source() -> types.Generator[int, None, None]: + for item in range(count): + consumed.append(item) + yield item + + with pytest.raises( + ValueError, match='batch_size must be greater than zero' + ): + list(python_utils.batcher(source(), batch_size)) + + assert consumed == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize('batch_size', [0, -1]) +@pytest.mark.parametrize('count', [0, 3]) +@pytest.mark.parametrize('interval', [None, 1]) +async def test_abatcher_rejects_nonpositive_size( + batch_size: int, count: int, interval: int | None +) -> None: + """Reject invalid sizes before scheduling a request from the source.""" + consumed: list[int] = [] + + async def source() -> types.AsyncGenerator[int, None]: + for item in range(count): + consumed.append(item) + yield item + + with pytest.raises( + ValueError, match='batch_size must be greater than zero' + ): + async for _ in python_utils.abatcher(source(), batch_size, interval): + pass + + assert consumed == [] + + 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..0794601 100644 --- a/python_utils/generators.py +++ b/python_utils/generators.py @@ -23,6 +23,12 @@ _T = typing.TypeVar('_T') +def _validate_batch_size(batch_size: int | None) -> None: + """Require a positive batch size, or None for interval-only batching.""" + if batch_size is not None and batch_size < 1: + raise ValueError('batch_size must be greater than zero') + + async def _cancel(pending: set[asyncio.Future[_T]]) -> None: """Cancel the items that are still on their way and wait for them. @@ -78,14 +84,19 @@ 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. Must be greater than zero when specified. Defaults to None. 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. + + Raises: + ValueError: If batch_size is specified but is not greater than zero. """ + _validate_batch_size(batch_size) + batch: list[_T] = [] assert batch_size or interval, 'Must specify either batch_size or interval' @@ -170,12 +181,17 @@ 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. Must be greater than zero. Yields: collections.abc.Generator[list[_T], None, None]: A generator that yields batches of items. + + Raises: + ValueError: If batch_size is not greater than zero. """ + _validate_batch_size(batch_size) + batch: list[_T] = [] for item in iterable: batch.append(item) From 3653324d7bb97d250c5690c747de922f14b18f05 Mon Sep 17 00:00:00 2001 From: tongwentao Date: Sun, 25 Oct 2026 23:58:17 +0800 Subject: [PATCH 2/3] fix: deprecate nonpositive batch sizes without breaking 4.x --- _python_utils_tests/test_generators.py | 97 +++++++++++++++++++++----- python_utils/generators.py | 39 ++++++++--- 2 files changed, 108 insertions(+), 28 deletions(-) diff --git a/_python_utils_tests/test_generators.py b/_python_utils_tests/test_generators.py index f0ca8cb..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( @@ -366,8 +367,10 @@ async def test_abatcher_with_future_returning_iterator() -> None: @pytest.mark.parametrize('batch_size', [0, -1]) @pytest.mark.parametrize('count', [0, 3]) -def test_batcher_rejects_nonpositive_size(batch_size: int, count: int) -> None: - """Reject invalid sizes without consuming even a finite source.""" +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]: @@ -375,22 +378,37 @@ def source() -> types.Generator[int, None, None]: consumed.append(item) yield item - with pytest.raises( - ValueError, match='batch_size must be greater than zero' - ): - list(python_utils.batcher(source(), batch_size)) - + 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, 1]) -async def test_abatcher_rejects_nonpositive_size( +@pytest.mark.parametrize('interval', [None, 0]) +async def test_abatcher_warns_without_interval( batch_size: int, count: int, interval: int | None ) -> None: - """Reject invalid sizes before scheduling a request from the source.""" + """Keep the old zero-size error and negative-size result in 4.x.""" consumed: list[int] = [] async def source() -> types.AsyncGenerator[int, None]: @@ -398,13 +416,58 @@ async def source() -> types.AsyncGenerator[int, None]: consumed.append(item) yield item - with pytest.raises( - ValueError, match='batch_size must be greater than zero' - ): - async for _ in python_utils.abatcher(source(), batch_size, interval): - pass - + 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: diff --git a/python_utils/generators.py b/python_utils/generators.py index 0794601..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,10 +24,20 @@ _T = typing.TypeVar('_T') -def _validate_batch_size(batch_size: int | None) -> None: - """Require a positive batch size, or None for interval-only batching.""" +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: - raise ValueError('batch_size must be greater than zero') + 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: @@ -84,7 +95,9 @@ async def abatcher( Args: generator: The async generator or iterator to batch. batch_size (typing.Optional[int], optional): The number of items per - batch. Must be greater than zero when specified. 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. @@ -92,10 +105,11 @@ async def abatcher( collections.abc.AsyncGenerator[list[_T], None]: A generator that yields batches of items. - Raises: - ValueError: If batch_size is specified but is not greater than zero. + Warns: + DeprecationWarning: If batch_size is below 1 without an interval. + The existing behavior is preserved until version 5.0. """ - _validate_batch_size(batch_size) + batch_size = _warn_nonpositive_batch_size(batch_size, interval) batch: list[_T] = [] @@ -181,16 +195,19 @@ 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. Must be greater than zero. + 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. - Raises: - ValueError: If batch_size is not greater than zero. + Warns: + DeprecationWarning: If batch_size is below 1. The warning is issued + when iteration begins, before consuming any source items. """ - _validate_batch_size(batch_size) + _warn_nonpositive_batch_size(batch_size) batch: list[_T] = [] for item in iterable: From 01916180ed3188fbd1cc128b634dd39665e41d41 Mon Sep 17 00:00:00 2001 From: tongwentao Date: Sun, 25 Oct 2026 23:58:17 +0800 Subject: [PATCH 3/3] docs: describe staged batch size deprecation --- CHANGELOG.md | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e474c71..9887c6e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,10 +2,11 @@ ## Unreleased -- Reject zero and negative batch sizes with `ValueError` in `batcher` and - `abatcher` before consuming the source. These sizes never reached the batch - limit and could buffer an unbounded source indefinitely. `abatcher` still - accepts `None` for interval-only batching. +- 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