diff --git a/src/monitor_utils/database.py b/src/monitor_utils/database.py index 4f8ca5f..16c6cec 100644 --- a/src/monitor_utils/database.py +++ b/src/monitor_utils/database.py @@ -217,7 +217,11 @@ def resolve_price_before_drop( previous_price on a drop, otherwise None. """ - if previous_price is not None and new_price is not None and new_price < previous_price: + if ( + previous_price is not None + and new_price is not None + and new_price < previous_price + ): return previous_price return None diff --git a/src/monitor_utils/runner.py b/src/monitor_utils/runner.py index d28412b..685db12 100644 --- a/src/monitor_utils/runner.py +++ b/src/monitor_utils/runner.py @@ -191,15 +191,15 @@ async def execute_monitor_logic( proxies: AioProxies, headers: dict | None = None, cookies: dict | None = None, - pre_task_func: Callable[..., Awaitable[None]] | None = None, + concurrent_pre_task: 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. - Each cycle opens a TLS or base session, optionally runs pre_task_func, - dispatches task_func per category concurrently, logs expected failures, - raises unexpected ones, then sleeps briefly before the next cycle. + Each cycle opens a TLS or base session, dispatches task_func per category + concurrently, logs expected failures, raises unexpected ones, then sleeps + briefly before the next cycle. Args: category_identifiers: List of category ids, or a mapping of category @@ -210,10 +210,13 @@ async def execute_monitor_logic( 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 before category - tasks, receiving session and **kwargs. + concurrent_pre_task: 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 and every task_func call. + **kwargs: Forwarded verbatim to concurrent_pre_task and every + task_func call. """ while True: @@ -227,15 +230,6 @@ async def execute_monitor_logic( ) async with session_ctx as session: LOGGER.info("starting tasks") - if pre_task_func: - try: - await pre_task_func(session=session, **kwargs) - except EXPECTED_TASK_ERRORS as exc: - LOGGER.warning("failed to get products", exc_info=exc) - except Exception: - LOGGER.exception("unexpected error") - raise - if isinstance(category_identifiers, dict): tasks = [ task_func(category_id, session, query_category_id=query, **kwargs) @@ -247,7 +241,11 @@ async def execute_monitor_logic( task_func(category_identifier, session, **kwargs) for category_identifier in category_identifiers ] - category_keys = category_identifiers + category_keys = list(category_identifiers) + + if concurrent_pre_task: + tasks.append(concurrent_pre_task(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): diff --git a/src/monitor_utils/sessions.py b/src/monitor_utils/sessions.py index 43a0689..b2e3181 100644 --- a/src/monitor_utils/sessions.py +++ b/src/monitor_utils/sessions.py @@ -146,6 +146,7 @@ async def send_request( 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: @@ -164,6 +165,8 @@ async def send_request( 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 @@ -193,6 +196,7 @@ async def send_request( params=params, json=json_data, data=data, + headers=headers, proxy=proxy, ssl=verify, ) as response: @@ -205,6 +209,7 @@ async def send_request( params=params, json=json_data, data=data, + headers=headers, proxy=proxy, verify=verify, )