small bug fix

This commit is contained in:
TPV committed 2026-07-16 14:19:33 -07:00
1 parent cdce46c033
commit 4aa0e43289
1 file changed
+17 -15
+17 -15
View File
@@ -200,9 +200,9 @@ async def execute_monitor_logic(
): ):
"""Run the monitor's category tasks in a loop with a fresh session per cycle. """Run the monitor's category tasks in a loop with a fresh session per cycle.
Each cycle opens a TLS or base session, dispatches task_func per category Each cycle opens a TLS or base session, awaits pre_task_func if given,
concurrently, logs expected failures, raises unexpected ones, then sleeps dispatches task_func per category concurrently, logs expected failures,
briefly before the next cycle. raises unexpected ones, then sleeps briefly before the next cycle.
Args: Args:
category_identifiers: List of category ids, or a mapping of category category_identifiers: List of category ids, or a mapping of category
@@ -223,8 +223,8 @@ async def execute_monitor_logic(
Category tasks must not depend on its side effects (session Category tasks must not depend on its side effects (session
headers, cookies, fetched state) since no ordering is guaranteed. headers, cookies, fetched state) since no ordering is guaranteed.
use_tls: Use the TLS fingerprint session instead of the base session. use_tls: Use the TLS fingerprint session instead of the base session.
**kwargs: Forwarded verbatim to concurrent_pre_task and every **kwargs: Forwarded verbatim to pre_task_func, concurrent_pre_task_func,
task_func call. and every task_func call.
""" """
while True: while True:
@@ -237,10 +237,21 @@ async def execute_monitor_logic(
else get_base_session(headers=headers, cookies=cookies, proxies=proxies) else get_base_session(headers=headers, cookies=cookies, proxies=proxies)
) )
async with session_ctx as session: 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
if not skip_cycle:
LOGGER.info("starting tasks") LOGGER.info("starting tasks")
if isinstance(category_identifiers, dict): if isinstance(category_identifiers, dict):
tasks = [ tasks = [
task_func(category_id, session, query_category_id=query, **kwargs) task_func(
category_id, session, query_category_id=query, **kwargs
)
for category_id, query in category_identifiers.items() for category_id, query in category_identifiers.items()
] ]
category_keys = list(category_identifiers.keys()) category_keys = list(category_identifiers.keys())
@@ -251,15 +262,6 @@ async def execute_monitor_logic(
] ]
category_keys = list(category_identifiers) category_keys = list(category_identifiers)
if pre_task_func:
try:
await pre_task_func(session=session, **kwargs)
except EXPECTED_TASK_ERRORS as error:
LOGGER.warning("Failed to get products", exc_info=error)
except Exception:
LOGGER.exception("Unexpected error")
raise
if concurrent_pre_task_func: if concurrent_pre_task_func:
tasks.append(concurrent_pre_task_func(session=session, **kwargs)) tasks.append(concurrent_pre_task_func(session=session, **kwargs))
category_keys.append("pre_task") category_keys.append("pre_task")