Merge pull request 'added optional semphore' (#5) from aioweb-migration into main

Reviewed-on: #5

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: Optional bound on concurrent page fetches. Omit it to let a
-            semaphore across all categories so the limit caps total fan-out; a
+            monitor with few pages fan out unbounded; pass one to cap monitors
-            per-call semaphore would instead bound each category in isolation,
+            with many pages. Pass a single shared semaphore across all categories
-            making the effective limit this value times the category count.
+            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."""
+        """Fetch a single page or offset, optionally under the shared semaphore."""
-        async with semaphore:
+        async with (semaphore or nullcontext()):
             if stop_event.is_set():
                 return
 

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']}"