migrate session handling to aioweb_tls/refactoring of helpers

src/monitor_utils/constants.py

@@ -11,6 +11,7 @@ from json import JSONDecodeError
 
 import aiohttp
 import requests
+from aioweb import AiowebError
 from curl_cffi.requests.errors import RequestsError
 from pymongo.errors import PyMongoError
 
@@ -27,6 +28,7 @@ LOGGERS = [
 
 RETRY_LIMIT = 5
 RETRY_WAIT = 3
+CYCLE_DELAY = 0.30
 MIN_BACKOFF_DELAY = 0.1
 MAX_BACKOFF_DELAY = 2.0
 FAILURE_DELAY = 3
@@ -44,6 +46,7 @@ EXPECTED_TASK_ERRORS = (
     requests.HTTPError,
     PyMongoError,
     JSONDecodeError,
+    AiowebError,
 )
 
 

src/monitor_utils/database.py

@@ -99,7 +99,8 @@ async def add_products_to_database(
     price is lower (a drop) and cleared otherwise, so any price_before_drop on
     the incoming product is ignored. Prices are recorded in a pastPrices array
     of price/date entries; a price already present in the array is never
-    recorded again, so each distinct price keeps only its first-seen timestamp. New products and never-before-recorded prices
+    recorded again, so each distinct price keeps only its first-seen timestamp.
+    New products and never-before-recorded prices
     upsert an alert referencing the product; a price returning to a value
     already in pastPrices emits nothing, so per-source price oscillation
     (member vs non-member quotes for the same SKU) alerts once per distinct
@@ -119,6 +120,12 @@ async def add_products_to_database(
             product_id.
     """
 
+    # Callers are expected to filter empty batches, but guard here too: an empty
+    # products list would make collection.bulk_write([]) raise InvalidOperation,
+    # which callers would then log as a spurious category failure every cycle.
+    if not products:
+        return
+
     doc_ids = [
         build_doc_id(p.product_id, category_identifier, use_composite) for p in products
     ]
@@ -311,8 +318,10 @@ def build_doc_id(
 def convert_product_to_mongo(product: Product) -> dict:
     """Flatten a Product into its MongoDB document representation.
 
-    Coerces string prices to floats and stamps lastUpdated with the current
-    UTC time. The _id is assigned by the caller via build_doc_id.
+    Stamps lastUpdated with the current UTC time. Prices are already floats via
+    the Product's pydantic validation (numeric strings coerced, comma strings
+    rejected at construction), so no coercion happens here. The _id is assigned
+    by the caller via build_doc_id.
 
     Args:
         product: Product to convert.
@@ -328,9 +337,6 @@ def convert_product_to_mongo(product: Product) -> dict:
     availability = data["availability"]
     current_price = pricing["current_price"]
     original_price = pricing["original_price"]
-    if isinstance(current_price, str):
-        current_price = float(current_price.replace(",", ""))
-        original_price = float(original_price.replace(",", ""))
 
     return {
         "title": data["title"],

src/monitor_utils/dsql_shortlinks.py

@@ -1,3 +1,4 @@
+# Idk who made this i was provided it by louis
 import asyncio
 from typing import Dict
 

src/monitor_utils/models.py

@@ -6,6 +6,24 @@ from datetime import datetime
 from pydantic.dataclasses import dataclass
 
 
+@dataclass
+class CycleResult:
+    """Tallied outcome of one monitor cycle.
+
+    Pure counts for the caller to log or act on as it sees fit; no policy is
+    baked in.
+
+    Attributes:
+        successes: Category tasks that completed without an expected error.
+        failures: Category tasks that raised an expected error.
+        skipped: True when pre_task_func failed and no category tasks ran.
+    """
+
+    successes: int = 0
+    failures: int = 0
+    skipped: bool = False
+
+
 @dataclass
 class ConsumerConfig:
     """Configuration for a single retailer's consumer.

src/monitor_utils/runner.py

@@ -1,6 +1,6 @@
 import asyncio
 import logging
-from typing import Awaitable, Callable, Iterable
+from collections.abc import Awaitable, Callable, Iterable
 
 import aiohttp
 from aioproxies import AioProxies
@@ -18,12 +18,12 @@ from monitor_utils.constants import EXPECTED_TASK_ERRORS, LOGGING_LEVEL
 from monitor_utils.database import get_collection
 from monitor_utils.models import (
     AvailabilityInfo,
+    CycleResult,
     PricingInfo,
     Product,
     ProductMetadata,
     ProductTiming,
 )
-from monitor_utils.sessions import get_base_session, get_tls_session
 
 LOGGER = logging.getLogger(__name__)
 
@@ -186,23 +186,23 @@ def build_product_details(
     return built
 
 
-async def execute_monitor_logic(
+async def run_monitor_cycle(
     category_identifiers: list[str] | dict[str, str],
     task_func: Callable[..., Awaitable[None]],
+    session: aiohttp.ClientSession | AsyncSession,
     *,
-    proxies: AioProxies,
-    headers: dict | None = None,
-    cookies: dict | None = None,
     pre_task_func: Callable[..., Awaitable[None]] | None = None,
     concurrent_pre_task_func: Callable[..., Awaitable[None]] | None = None,
-    use_tls: bool = False,
     **kwargs,
-):
-    """Run the monitor's category tasks in a loop with a fresh session per cycle.
+) -> CycleResult:
+    """Run one monitor cycle over a caller-supplied session and report the outcome.
 
-    Each cycle opens a TLS or base session, awaits pre_task_func if given,
-    dispatches task_func per category concurrently, logs expected failures,
-    raises unexpected ones, then sleeps briefly before the next cycle.
+    Awaits pre_task_func if given, dispatches task_func per category
+    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).
 
     Args:
         category_identifiers: List of category ids, or a mapping of category
@@ -210,80 +210,78 @@ async def execute_monitor_logic(
             task_func.
         task_func: Awaitable invoked per category with the identifier,
             session, and **kwargs.
-        proxies: Proxies applied to the session.
-        headers: Headers applied to the session.
-        cookies: Cookies applied to the base session; ignored for TLS.
-        pre_task_func: Optional awaitable run once per 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 retried after the usual delay.
+        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.
+        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.
         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.
-        use_tls: Use the TLS fingerprint session instead of the base session.
         **kwargs: Forwarded verbatim to pre_task_func, concurrent_pre_task_func,
             and every task_func call.
+
+    Returns:
+        A CycleResult tallying task successes and expected-error failures, or a
+        skipped result when pre_task_func failed.
+
+    Raises:
+        Exception: Any unexpected (non-EXPECTED_TASK_ERRORS) error raised by a
+            task or pre_task_func propagates, failing loud so the caller's loop
+            crashes rather than silently retrying a broken monitor.
     """
 
-    while True:
-        session_ctx = (
-            get_tls_session(
-                headers=headers,
-                proxies=proxies,
+    if pre_task_func:
+        try:
+            await pre_task_func(session=session, **kwargs)
+        except EXPECTED_TASK_ERRORS as error:
+            LOGGER.warning("pre task failed, skipping cycle", exc_info=error)
+            return CycleResult(skipped=True)
+
+    LOGGER.info("starting tasks")
+    if isinstance(category_identifiers, dict):
+        tasks = [
+            task_func(category_id, session, 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)
+            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))
+        category_keys.append("pre_task")
+
+    results = await asyncio.gather(*tasks, return_exceptions=True)
+    successes = 0
+    failures = 0
+    for category_identifier, result in zip(category_keys, results):
+        if isinstance(result, EXPECTED_TASK_ERRORS):
+            LOGGER.warning(
+                "failed to get products",
+                extra={"category": category_identifier},
+                exc_info=result,
             )
-            if use_tls
-            else get_base_session(headers=headers, cookies=cookies, proxies=proxies)
-        )
-        async with session_ctx as session:
-            skip_cycle = False
-            if pre_task_func:
-                try:
-                    await pre_task_func(session=session, **kwargs)
-                except EXPECTED_TASK_ERRORS as error:
-                    LOGGER.warning("pre task failed, skipping cycle", exc_info=error)
-                    skip_cycle = True
+            failures += 1
+        elif isinstance(result, Exception):
+            LOGGER.error(
+                "unexpected error",
+                extra={"category": category_identifier},
+                exc_info=result,
+            )
+            raise result
+        else:
+            successes += 1
 
-            if not skip_cycle:
-                LOGGER.info("starting tasks")
-                if isinstance(category_identifiers, dict):
-                    tasks = [
-                        task_func(
-                            category_id, session, 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)
-                        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))
-                    category_keys.append("pre_task")
-
-                results = await asyncio.gather(*tasks, return_exceptions=True)
-                for category_identifier, result in zip(category_keys, results):
-                    if isinstance(result, EXPECTED_TASK_ERRORS):
-                        LOGGER.warning(
-                            "failed to get products",
-                            extra={"category": category_identifier},
-                            exc_info=result,
-                        )
-                    elif isinstance(result, Exception):
-                        LOGGER.error(
-                            "unexpected error",
-                            extra={"category": category_identifier},
-                            exc_info=result,
-                        )
-                        raise result
-
-        LOGGER.info("sleeping before next tasks", extra={"delay": 0.30})
-        await asyncio.sleep(0.30)
+    return CycleResult(successes=successes, failures=failures)
 
 
 async def execute_pagination(
@@ -311,11 +309,10 @@ 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.
-        max_concurrent_tasks: Semaphore limit for concurrent page fetches, used
-            only when semaphore is not supplied. Each execute_pagination call
-            builds its own semaphore, so this bounds one category in isolation;
-            with many categories running concurrently the effective limit is
-            this value times the category count.
+        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.
         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.

src/monitor_utils/sessions.py

@@ -1,225 +1,89 @@
 """Shared HTTP session helpers for all monitors.
 
-Known gap: UA/TLS fingerprint does not always match headers/header order,
-but targeted antibots do not currently seem to care.
+A single aioweb_tls TLSSession (curl_cffi backend, random client fingerprint per
+session) replaces the old aiohttp/curl_cffi split. send_request wraps aioweb's
+request_with_retries, adding per-request proxy rotation and old-style JSON/text
+return values so monitor call sites change as little as possible.
 """
 
-import asyncio
-import json
-import logging
 import random
-from contextlib import asynccontextmanager
-from typing import AsyncGenerator
 
-import aiohttp
-import requests
 from aioproxies import AioProxies
-from curl_cffi.requests import AsyncSession
-from curl_cffi.requests.errors import RequestsError
-from curl_cffi.requests.exceptions import Timeout as CurlTimeout
-from fake_useragent import FakeUserAgent
-from tenacity import (
-    after_log,
-    before_log,
-    retry,
-    retry_if_exception_type,
-    stop_after_attempt,
-    wait_fixed,
-)
+from aioweb import Response
+from aioweb_tls import CurlCffi, TLSSession
 
-from monitor_utils.constants import (
-    CLIENT_IDENTIFIERS,
-    CLIENT_TIMEOUT,
-    RETRY_LIMIT,
-    RETRY_WAIT,
-    TIMEOUT_SECONDS,
-)
-
-LOGGER = logging.getLogger(__name__)
-UA = FakeUserAgent()
+from monitor_utils.constants import CLIENT_IDENTIFIERS, RETRY_LIMIT
 
 
-@asynccontextmanager
-async def get_base_session(
-    headers: dict | None = None,
-    cookies: dict | None = None,
-    connector: aiohttp.TCPConnector | None = None,
-    timeout: aiohttp.ClientTimeout = CLIENT_TIMEOUT,
-    proxies: AioProxies | None = None,
-) -> AsyncGenerator[aiohttp.ClientSession, None]:
-    """Yield an aiohttp session with a random User-Agent and attached proxy pool.
+def get_default_session() -> TLSSession:
+    """Return a TLSSession impersonating a random client fingerprint.
 
-    The proxy pool is stored on the session as proxy_pool for send_request to
-    rotate per request. The session is always closed on exit.
+    Fingerprint rotation lives here so every monitor rotates the same way; the
+    caller owns the session lifetime via ``async with``.
 
-    Args:
-        headers: Headers applied to the session; User-Agent is defaulted if
-            absent.
-        cookies: Cookies applied to the session.
-        connector: Optional TCP connector.
-        timeout: Client timeout applied to the session.
-        proxies: Proxy pool attached to the session.
-
-    Yields:
-        The configured aiohttp session.
+    Returns:
+        A TLSSession over a curl_cffi backend with a random impersonation.
     """
 
-    headers = headers if headers else {}
-    headers.setdefault("User-Agent", UA.random)
-    session = aiohttp.ClientSession(
-        headers=headers, cookies=cookies, connector=connector, timeout=timeout
-    )
-    session.proxy_pool = proxies
-    try:
-        yield session
-    finally:
-        await session.close()
+    return TLSSession(backend=CurlCffi(impersonate=random.choice(CLIENT_IDENTIFIERS)))
 
 
-@asynccontextmanager
-async def get_tls_session(
-    headers: dict | None = None,
-    timeout: int = TIMEOUT_SECONDS,
-    proxies: AioProxies | None = None,
-) -> AsyncGenerator[AsyncSession, None]:
-    """Yield a curl_cffi session impersonating a random client fingerprint.
-
-    The proxy pool is stored on the session as proxy_pool for send_request to
-    rotate per request.
-
-    Args:
-        headers: Headers applied to the session; a Chrome User-Agent is
-            defaulted if absent.
-        timeout: Request timeout in seconds.
-        proxies: Proxy pool attached to the session.
-
-    Yields:
-        The configured curl_cffi session.
-    """
-
-    headers = headers if headers else {}
-    headers.setdefault("User-Agent", UA.chrome)
-    async with AsyncSession(
-        impersonate=random.choice(CLIENT_IDENTIFIERS),
-        timeout=timeout,
-        headers=headers,
-    ) as session:
-        session.proxy_pool = proxies
-        yield session
-
-
-def build_session_details(
-    session: aiohttp.ClientSession | AsyncSession, headers: dict = None
-):
-    """Merge headers into an existing session, defaulting a random User-Agent.
-
-    Args:
-        session: Session whose headers are updated in place.
-        headers: Headers to merge; User-Agent is defaulted if absent.
-    """
-
-    headers = headers or {}
-    headers.setdefault("User-Agent", UA.random)
-    session.headers.update(headers)
-
-
-@retry(
-    stop=stop_after_attempt(RETRY_LIMIT),
-    wait=wait_fixed(RETRY_WAIT),
-    retry=retry_if_exception_type(
-        (
-            aiohttp.ClientError,
-            asyncio.TimeoutError,
-            RequestsError,
-            requests.Timeout,
-            CurlTimeout,
-        )
-    ),
-    before=before_log(LOGGER, logging.DEBUG),
-    after=after_log(LOGGER, logging.DEBUG),
-    reraise=True,
-)
 async def send_request(
-    session: aiohttp.ClientSession | AsyncSession,
+    session: TLSSession,
     url: str,
     method: str,
+    *,
+    proxies: AioProxies,
     params: dict | None = None,
     json_data: dict | None = None,
     data: dict | None = None,
     headers: dict | None = None,
-    reraise_json_error: bool = False,
-    development_proxy: str | None = None,
-) -> dict | str:
-    """Send a request through either session type with retries and proxy rotation.
+) -> Response:
+    """Send a request through a TLSSession with retries and proxy rotation.
 
-    Retries transport errors and timeouts up to RETRY_LIMIT with a fixed wait.
-    When development_proxy is set it overrides the session's proxy pool and
-    disables TLS verification for local traffic inspection; otherwise the next
-    proxy is drawn from the session's proxy_pool when present. Responses are
-    parsed as JSON when possible.
+    Wraps aioweb's request_with_retries, which retries transport errors,
+    timeouts, and retryable statuses (429/5xx) up to RETRY_LIMIT attempts with
+    exponential backoff. The next proxy is drawn from the pool per call and
+    applied to both schemes. A non-2xx response or a fully failed request raises
+    AiowebError, an expected task error, so a failed fetch surfaces the same way
+    the old HTTPError path did. On success the aioweb Response is returned for
+    the caller to read via response.json(), response.text(), etc.
 
     Args:
-        session: aiohttp or curl_cffi session to send through.
+        session: The TLSSession to send through.
         url: Request URL.
         method: HTTP method.
+        proxies: Proxy pool (required); the next proxy is drawn per call and
+            applied to both http and https.
         params: Query parameters.
-        json_data: JSON body.
-        data: Form body.
-        headers: Per-request headers merged over the session headers,
-            avoiding mutation of shared session state from concurrent tasks.
-        reraise_json_error: Raise JSONDecodeError instead of returning raw
-            text when the response is not valid JSON.
-        development_proxy: Local proxy URL that overrides the pool and
-            disables verification.
+        json_data: JSON body. A dict here is sent as a JSON body.
+        data: Alternative body. Note aioweb routes any dict or list to a JSON
+            body, so true form-encoding requires a pre-encoded string; a form
+            dict passed here is sent as JSON.
+        headers: Per-request headers merged over the session defaults.
 
     Returns:
-        Parsed JSON dict, or raw response text when parsing fails and
-        reraise_json_error is False.
+        The aioweb Response, guaranteed 2xx (non-2xx raises). Read the body with
+        response.json() (dict/list, or None if not valid JSON) or
+        response.text().
 
     Raises:
-        json.JSONDecodeError: When parsing fails and reraise_json_error is
-            True.
+        AiowebError: On a non-2xx response or after all retries fail.
     """
 
-    verify = True
-    if development_proxy:
-        proxy = development_proxy
-        verify = False
-    else:
-        manager = getattr(session, "proxy_pool", None)
-        proxy = manager.next().url() if manager else None
+    proxy_url = proxies.next().url()
+    proxy = {"http": proxy_url, "https": proxy_url}
 
-    if isinstance(session, aiohttp.ClientSession):
-        async with session.request(
-            method=method,
-            url=url,
-            params=params,
-            json=json_data,
-            data=data,
-            headers=headers,
-            proxy=proxy,
-            ssl=verify,
-        ) as response:
-            response.raise_for_status()
-            text = await response.text()
-    else:
-        response = await session.request(
-            method=method.upper(),
-            url=url,
-            params=params,
-            json=json_data,
-            data=data,
-            headers=headers,
-            proxy=proxy,
-            verify=verify,
-        )
-        response.raise_for_status()
-        text = response.text
+    body = json_data if json_data is not None else data
+    response = await session.request_with_retries(
+        method.upper(),
+        url,
+        params=params,
+        data=body,
+        headers=headers,
+        proxies=proxy,
+        attempts=RETRY_LIMIT,
+    )
+    response.raise_for_status()
 
-    try:
-        return json.loads(text)
-    except json.JSONDecodeError:
-        if reraise_json_error:
-            raise
-
-        return text
+    return response