Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9af7e7adcf | ||
|
|
4b63a2d39c |
No files matched your search
@@ -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
|
||||
|
||||
|
||||
@@ -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']}"
|
||||
Reference in new issue
Block a user