diff --git a/samsung_shop/samsung_shop_monitor.py b/samsung_shop/samsung_shop_monitor.py index 5b2fdb6..c07ff5c 100644 --- a/samsung_shop/samsung_shop_monitor.py +++ b/samsung_shop/samsung_shop_monitor.py @@ -8,12 +8,13 @@ from monitor_utils.constants import CYCLE_DELAY, EXPECTED_TASK_ERRORS from monitor_utils.database import add_products_to_database from monitor_utils.models import AvailabilityInfo, PricingInfo, Product, ProductMetadata from monitor_utils.runner import ( - run_monitor_cycle, execute_pagination, + fan_out_group, init_monitor, + run_bounded_cycle, setup_monitor_logging, ) -from monitor_utils.sessions import get_default_session, send_request +from monitor_utils.sessions import send_request from motor.motor_asyncio import AsyncIOMotorCollection setup_monitor_logging(monitor_name="samsung_shop") @@ -26,20 +27,19 @@ CATEGORY_SEMAPHORE = asyncio.Semaphore(CATEGORY_CONCURRENCY) async def main(): - """Run the samsung shop monitor loop over all category ids using a base session.""" + """Run the samsung shop monitor loop over all category ids on bounded cycles.""" category_paths, collection, proxies = await init_monitor( monitor_name="samsung_shop", development_mode=DEVELOPMENT_MODE ) - async with get_default_session(proxies) as session: - while True: - await run_monitor_cycle( - category_paths, - task_func=get_category_products, - session=session, - collection=collection, - ) - LOGGER.info("tasks completed sleeping", extra={"delay": CYCLE_DELAY}) - await asyncio.sleep(CYCLE_DELAY) + while True: + await run_bounded_cycle( + category_paths, + get_category_products, + proxies, + collection, + ) + LOGGER.info("tasks completed sleeping", extra={"delay": CYCLE_DELAY}) + await asyncio.sleep(CYCLE_DELAY) async def get_category_products( @@ -81,8 +81,10 @@ async def get_category_products( await process_page(data, category_id, offset, collection) return - page_task = asyncio.create_task(process_page(data, category_id, offset, collection)) - try: + # offset zero owns the fan-out: the group binds this page's processing to the + # pagination it spawns, so a cancelled cycle takes both down together. + async with fan_out_group() as group: + group.create_task(process_page(data, category_id, offset, collection)) total_products = data.get("searchTotalCount") if total_products is None: LOGGER.warning("missing searchTotalCount", extra={"category": category_id}) @@ -98,8 +100,6 @@ async def get_category_products( category_id=category_id, collection=collection, ) - finally: - await page_task async def process_page(