src/monitor_utils/runner.py

@@ -190,6 +190,7 @@ async def run_monitor_cycle(
     category_identifiers: list[str] | dict[str, str],
     task_func: Callable[..., Awaitable[None]],
     session: aiohttp.ClientSession | AsyncSession,
+    collection: AsyncIOMotorCollection,
     *,
     pre_task_func: Callable[..., Awaitable[None]] | None = None,
     concurrent_pre_task_func: Callable[..., Awaitable[None]] | None = None,
@@ -201,27 +202,36 @@ async def run_monitor_cycle(
     concurrently, logs expected failures, and raises unexpected ones. The
     caller owns the loop, the session lifetime, and the inter-cycle delay: wrap
     this in a `while True` that opens a fresh session per cycle and sleeps
-    between cycles, using the returned CycleResult to drive backoff (reset the
-    delay when healthy, grow it when not).
+    between cycles. The returned CycleResult is pure counts for the caller to
+    log or act on; this function applies no policy to it.
+
+    session and collection are the two per-cycle resources, passed to every
+    callback (pre_task_func, concurrent_pre_task_func, and each task_func) just
+    like each other. A callback that does not need one still receives it, so
+    every callback must accept **kwargs to absorb the ones it ignores.
 
     Args:
         category_identifiers: List of category ids, or a mapping of category
             id to query id; mapping entries pass query_category_id to
             task_func.
-        task_func: Awaitable invoked per category with the identifier,
-            session, and **kwargs.
+        task_func: Awaitable invoked per category with the identifier, session,
+            collection, and **kwargs.
         session: Open HTTP session for this cycle, built and closed by the
-            caller. Proxies are the caller's responsibility, either bound to
-            the session or forwarded to task_func via **kwargs.
+            caller. Proxies are the caller's responsibility, bound to the
+            session (see get_default_session).
+        collection: Mongo collection the tasks persist products to; forwarded to
+            every callback so any that writes has it.
         pre_task_func: Optional awaitable run once at the start of the cycle,
             awaited before any category tasks are dispatched and receiving
-            session and **kwargs. Use it for setup the category tasks depend on
-            (session headers, cookies, fetched state); if it raises an expected
-            task error the cycle is skipped and a skipped CycleResult returned.
+            session, collection, and **kwargs. Use it for setup the category
+            tasks depend on (session headers, cookies, fetched state); if it
+            raises an expected task error the cycle is skipped and a skipped
+            CycleResult returned.
         concurrent_pre_task_func: Optional awaitable run once per cycle, gathered
-            alongside the category tasks and receiving session and **kwargs.
-            Category tasks must not depend on its side effects (session
-            headers, cookies, fetched state) since no ordering is guaranteed.
+            alongside the category tasks and receiving session, collection, and
+            **kwargs. Category tasks must not depend on its side effects
+            (session headers, cookies, fetched state) since no ordering is
+            guaranteed.
         **kwargs: Forwarded verbatim to pre_task_func, concurrent_pre_task_func,
             and every task_func call.
 
@@ -237,7 +247,7 @@ async def run_monitor_cycle(
 
     if pre_task_func:
         try:
-            await pre_task_func(session=session, **kwargs)
+            await pre_task_func(session=session, collection=collection, **kwargs)
         except EXPECTED_TASK_ERRORS as error:
             LOGGER.warning("pre task failed, skipping cycle", exc_info=error)
             return CycleResult(skipped=True)
@@ -245,19 +255,23 @@ async def run_monitor_cycle(
     LOGGER.info("starting tasks")
     if isinstance(category_identifiers, dict):
         tasks = [
-            task_func(category_id, session, query_category_id=query, **kwargs)
+            task_func(
+                category_id, session, collection, query_category_id=query, **kwargs
+            )
             for category_id, query in category_identifiers.items()
         ]
         category_keys = list(category_identifiers.keys())
     else:
         tasks = [
-            task_func(category_identifier, session, **kwargs)
+            task_func(category_identifier, session, collection, **kwargs)
             for category_identifier in category_identifiers
         ]
         category_keys = list(category_identifiers)
 
     if concurrent_pre_task_func:
-        tasks.append(concurrent_pre_task_func(session=session, **kwargs))
+        tasks.append(
+            concurrent_pre_task_func(session=session, collection=collection, **kwargs)
+        )
         category_keys.append("pre_task")
 
     results = await asyncio.gather(*tasks, return_exceptions=True)