From 4b63a2d39c89461f880db872fdef6397aee91147 Mon Sep 17 00:00:00 2001 From: TPV Date: Mon, 20 Jul 2026 19:27:47 -0700 Subject: [PATCH] added optional semphore --- src/monitor_utils/runner.py | 17 ++++++++++------- tests/test_pagination_semaphore.py | 23 +++++++++++++++++++++++ 2 files changed, 33 insertions(+), 7 deletions(-) diff --git a/src/monitor_utils/runner.py b/src/monitor_utils/runner.py index 27ef77d..a9d528e 100644 --- a/src/monitor_utils/runner.py +++ b/src/monitor_utils/runner.py @@ -1,6 +1,7 @@ import asyncio import logging from collections.abc import Awaitable, Callable, Iterable +from contextlib import nullcontext import aiohttp from aioproxies import AioProxies @@ -305,7 +306,7 @@ async def execute_pagination( func: Callable[..., Awaitable[None]], session: aiohttp.ClientSession | AsyncSession, *, - semaphore: asyncio.Semaphore, + semaphore: asyncio.Semaphore | None = None, start_offset: int = 0, pagination_mode: str = "offset", **kwargs, @@ -323,10 +324,12 @@ async def execute_pagination( func: Awaitable task invoked per page; receives session plus either offset or page_number depending on pagination_mode. session: HTTP session forwarded to each func call. - semaphore: Required bound on concurrent page fetches. Pass one shared - semaphore across all categories so the limit caps total fan-out; a - per-call semaphore would instead bound each category in isolation, - making the effective limit this value times the category count. + semaphore: Optional bound on concurrent page fetches. Omit it to let a + monitor with few pages fan out unbounded; pass one to cap monitors + with many pages. Pass a single shared semaphore across all categories + so the limit caps total fan-out; a per-call semaphore would instead + bound each category in isolation, making the effective limit this + value times the category count. start_offset: Additive base applied to every offset, for APIs whose offset space does not begin at zero. Page N requests offset start_offset + (N - 1) * products_per_page. @@ -347,8 +350,8 @@ async def execute_pagination( page_range = range(2, total_pages + 1) async def fetch_page(page_index: int): - """Fetch a single page or offset under the shared semaphore.""" - async with semaphore: + """Fetch a single page or offset, optionally under the shared semaphore.""" + async with (semaphore or nullcontext()): if stop_event.is_set(): return diff --git a/tests/test_pagination_semaphore.py b/tests/test_pagination_semaphore.py index ec99241..a737d03 100644 --- a/tests/test_pagination_semaphore.py +++ b/tests/test_pagination_semaphore.py @@ -82,3 +82,26 @@ def test_single_call_respects_its_semaphore(): asyncio.run(run()) assert state["peak"] <= limit, f"peak {state['peak']} over limit {limit}" + + +def test_omitted_semaphore_runs_unbounded(): + """Omitting the semaphore lets all pages fan out concurrently. + + With no semaphore the only bound is the page count; pages 2..10 (nine pages) + should all be in flight at once, so peak concurrency reaches nine. + """ + + state = {"live": 0, "peak": 0} + func = _make_tracking_func(state) + + async def run(): + await execute_pagination( + "cat", + total_products=300, + products_per_page=30, + func=func, + session=None, + ) + + asyncio.run(run()) + assert state["peak"] == 9, f"expected all 9 pages concurrent, got {state['peak']}" -- 2.54.0