Compare commits

...
2 Commits
Author SHA1 Message Date
TPV 9af7e7adcf Merge pull request 'added optional semphore' (#5) from aioweb-migration into main
Reviewed-on: #5
2026-07-21 03:05:50 +00:00
TPV 4b63a2d39c added optional semphore 2026-07-20 19:27:47 -07:00
2 changed files with 33 additions and 7 deletions

No files matched your search

+10 -7
View File
@@ -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
+23
View File
@@ -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']}"