680 lines
25 KiB
Python
680 lines
25 KiB
Python
import asyncio
|
|
import logging
|
|
from collections.abc import Awaitable, Callable, Iterable
|
|
from contextlib import asynccontextmanager, nullcontext
|
|
|
|
import aiohttp
|
|
from aioproxies import AioProxies
|
|
from curl_cffi.requests import AsyncSession
|
|
from log_setup import setup_logging
|
|
from motor.motor_asyncio import AsyncIOMotorCollection
|
|
|
|
from monitor_utils.config import (
|
|
DevelopmentMode,
|
|
get_config,
|
|
is_production,
|
|
set_logger_levels,
|
|
)
|
|
from monitor_utils.constants import (
|
|
CYCLE_TIMEOUT_SECONDS,
|
|
EXPECTED_TASK_ERRORS,
|
|
LOGGING_LEVEL,
|
|
MAX_CURL_HANDLES,
|
|
TIMEOUT_SECONDS,
|
|
)
|
|
from monitor_utils.database import get_collection
|
|
from monitor_utils.sessions import get_default_session
|
|
from monitor_utils.models import (
|
|
AvailabilityInfo,
|
|
CycleResult,
|
|
PricingInfo,
|
|
Product,
|
|
ProductMetadata,
|
|
ProductTiming,
|
|
)
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
def setup_monitor_logging(monitor_name: str, logging_level=LOGGING_LEVEL):
|
|
"""Configure JSON logging for a monitor and quiet noisy libraries.
|
|
|
|
Call once at module import, at global scope, before creating module loggers,
|
|
so logging is configured synchronously at startup rather than inside the
|
|
async init_monitor. Logs go to logs/latest.log with rotated history keyed by
|
|
monitor_name.
|
|
|
|
Args:
|
|
monitor_name: Monitor name used as the rotated-history base.
|
|
logging_level: Root logging level.
|
|
"""
|
|
|
|
setup_logging(
|
|
name="logs/latest.log",
|
|
history_name=monitor_name,
|
|
level=logging_level,
|
|
output="json",
|
|
)
|
|
set_logger_levels()
|
|
|
|
|
|
async def init_monitor(
|
|
monitor_name: str,
|
|
development_mode: DevelopmentMode,
|
|
production_proxy_path: str = "proxies.txt",
|
|
testing_proxy_path: str = "tests/configs/proxies.txt",
|
|
) -> tuple[list[str] | dict[str, str], AsyncIOMotorCollection, AioProxies]:
|
|
"""Load a monitor's config, collection, and proxies for the mode.
|
|
|
|
Async because config now lives in Mongo; call it from inside the monitor's
|
|
event loop (in main), not at module import. Logging is configured separately
|
|
by setup_monitor_logging at module scope and is not touched here.
|
|
|
|
Args:
|
|
monitor_name: Monitor identifier used for config and collection lookup.
|
|
development_mode: Selects production or testing config, collection,
|
|
and proxy path.
|
|
production_proxy_path: Proxy file path used in production mode.
|
|
testing_proxy_path: Proxy file path used in testing mode.
|
|
|
|
Returns:
|
|
Tuple of category names, Mongo collection, and loaded proxies.
|
|
"""
|
|
|
|
use_production = is_production(development_mode)
|
|
category_names, _, collection = await setup_config(
|
|
monitor_name=monitor_name, development_mode=development_mode
|
|
)
|
|
proxy_path = production_proxy_path if use_production else testing_proxy_path
|
|
proxies = AioProxies.from_file(path=proxy_path)
|
|
return category_names, collection, proxies
|
|
|
|
|
|
async def setup_config(
|
|
monitor_name: str, development_mode: DevelopmentMode
|
|
) -> tuple[
|
|
list[str] | dict[str, str], dict[str, str | list[str]], AsyncIOMotorCollection
|
|
]:
|
|
"""Load monitor config and resolve the Mongo collection for the mode.
|
|
|
|
Testing mode reads the configs_testing collection and appends a _testing
|
|
suffix to the product collection name so production data is never touched
|
|
during development.
|
|
|
|
Args:
|
|
monitor_name: Monitor identifier used for config and collection lookup.
|
|
development_mode: Selects the production or testing config and
|
|
collection.
|
|
|
|
Returns:
|
|
Tuple of category names, channel id mapping, and Mongo collection.
|
|
"""
|
|
|
|
use_production = is_production(development_mode)
|
|
config = await get_config(
|
|
monitor_name=monitor_name,
|
|
use_production=use_production,
|
|
)
|
|
|
|
category_names = config["categories"]
|
|
collection = get_collection(
|
|
monitor_name if use_production else f"{monitor_name}_testing"
|
|
)
|
|
return category_names, config["channels"], collection
|
|
|
|
|
|
def combined_discount(base_percentage: int, promo_percentage: int | None) -> int:
|
|
"""Combine an on-sale discount with a percentage code applied at checkout.
|
|
|
|
A "extra N% off" code discounts the already-reduced sale price, so the two
|
|
compound rather than add: 20% off then an extra 30% code is 44% off list,
|
|
not 50%. Returns base_percentage unchanged when there is no percentage code.
|
|
|
|
Args:
|
|
base_percentage: The on-sale discount off list, as a whole percent.
|
|
promo_percentage: The code's advertised percent off, or None.
|
|
|
|
Returns:
|
|
The combined discount off list, rounded to a whole percent.
|
|
"""
|
|
|
|
if not promo_percentage:
|
|
return base_percentage
|
|
|
|
return round(100 - (100 - base_percentage) * (100 - promo_percentage) / 100)
|
|
|
|
|
|
def build_product_details(
|
|
product: dict,
|
|
min_percentage: int,
|
|
min_price_drop: float | None = None,
|
|
) -> Product | None:
|
|
"""Build a Product from a Mongo document if it passes discount thresholds.
|
|
|
|
Skips documents with a missing or zero original price. Passing requires
|
|
meeting min_percentage, or min_price_drop when provided. Channel routing
|
|
is owned by the consumer bot; no channel id is attached here.
|
|
|
|
Args:
|
|
product: Mongo document for the product.
|
|
min_percentage: Minimum discount percentage to pass.
|
|
min_price_drop: Optional minimum absolute price drop to pass.
|
|
|
|
Returns:
|
|
The built Product, or None when skipped or below thresholds.
|
|
"""
|
|
|
|
product_id = product["_id"]
|
|
current_price = product["currentPrice"]
|
|
original_price = product["originalPrice"]
|
|
if not original_price:
|
|
LOGGER.warning(
|
|
"missing or zero original price, skipping",
|
|
extra={"product_id": product_id},
|
|
)
|
|
return
|
|
|
|
percentage = int(((original_price - current_price) / original_price) * 100)
|
|
price_drop = original_price - current_price
|
|
# A percentage code compounds onto the sale price, so an item can clear the
|
|
# threshold on the combined discount even when the base discount would not.
|
|
# Dollar codes do not map to a percent and never affect the threshold.
|
|
effective_percentage = combined_discount(percentage, product.get("promoPercentage"))
|
|
passes_percentage = effective_percentage >= min_percentage
|
|
passes_price_drop = min_price_drop is not None and price_drop >= min_price_drop
|
|
if not passes_percentage and not passes_price_drop:
|
|
return
|
|
|
|
built = Product(
|
|
product_id=product_id,
|
|
title=product["title"],
|
|
url=product["url"],
|
|
metadata=ProductMetadata(
|
|
image_url=product["imageUrl"],
|
|
customization_index=product["customizationIndex"],
|
|
member_non_member_comparison=product["memberNonMemberComparison"],
|
|
overstock_text=product["overstockText"],
|
|
percentage=percentage,
|
|
affiliate_url=product.get("affiliateUrl"),
|
|
promo_code=product.get("promoCode"),
|
|
promo_percentage=product.get("promoPercentage"),
|
|
promo_dollar_off=product.get("promoDollarOff"),
|
|
promo_min_spend=product.get("promoMinSpend"),
|
|
),
|
|
timing=ProductTiming(
|
|
discovered_date=product["discoveredDate"],
|
|
start_date=product["startDate"],
|
|
end_date=product["endDate"],
|
|
),
|
|
pricing=PricingInfo(
|
|
current_price=current_price,
|
|
original_price=original_price,
|
|
price_before_drop=product["priceBeforeDrop"],
|
|
edu_price=product["eduPrice"],
|
|
past_prices=product["pastPrices"],
|
|
),
|
|
availability=AvailabilityInfo(
|
|
available=product["available"],
|
|
condition=product["condition"],
|
|
app_exclusive=product["appExclusive"],
|
|
member_pick=product["memberPick"],
|
|
member_deal=product["memberDeal"],
|
|
),
|
|
)
|
|
return built
|
|
|
|
|
|
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,
|
|
**kwargs,
|
|
) -> CycleResult:
|
|
"""Run one monitor cycle over a caller-supplied session and report the outcome.
|
|
|
|
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 — or use run_bounded_cycle, which owns the session and a
|
|
hard cycle budget and leaves only the loop and the sleep to the caller.
|
|
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,
|
|
collection, and **kwargs.
|
|
session: Open HTTP session for this cycle, built and closed by the
|
|
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, 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, 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.
|
|
|
|
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.
|
|
"""
|
|
|
|
if pre_task_func:
|
|
try:
|
|
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)
|
|
|
|
LOGGER.info("starting tasks")
|
|
if isinstance(category_identifiers, dict):
|
|
tasks = [
|
|
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, 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, collection=collection, **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,
|
|
)
|
|
failures += 1
|
|
elif isinstance(result, Exception):
|
|
LOGGER.error(
|
|
"unexpected error",
|
|
extra={"category": category_identifier},
|
|
exc_info=result,
|
|
)
|
|
raise result
|
|
else:
|
|
successes += 1
|
|
|
|
return CycleResult(successes=successes, failures=failures)
|
|
|
|
|
|
async def run_bounded_cycle(
|
|
category_identifiers: list[str] | dict[str, str],
|
|
task_func: Callable[..., Awaitable[None]],
|
|
proxies: AioProxies,
|
|
collection: AsyncIOMotorCollection,
|
|
*,
|
|
headers: dict | None = None,
|
|
session_timeout: int = TIMEOUT_SECONDS,
|
|
max_clients: int = MAX_CURL_HANDLES,
|
|
cycle_timeout: int = CYCLE_TIMEOUT_SECONDS,
|
|
**kwargs,
|
|
) -> CycleResult | None:
|
|
"""Run one monitor cycle on a fresh session under a hard time budget.
|
|
|
|
The session is opened and closed inside the call, so curl multi-handle
|
|
state cannot outlive one cycle: a wedged transfer dies with its session
|
|
instead of parking the process. The budget turns an engine-level lost
|
|
wakeup into a caught TimeoutError and an abandoned cycle, never a
|
|
permanent wedge; the session's own close is bounded by
|
|
get_default_session, so the exit cannot park past the budget either.
|
|
|
|
The caller still owns the loop and the inter-cycle delay:
|
|
|
|
while True:
|
|
await run_bounded_cycle(ids, task_func, proxies, collection)
|
|
await asyncio.sleep(CYCLE_DELAY)
|
|
|
|
Args:
|
|
category_identifiers: Categories, passed through to run_monitor_cycle.
|
|
task_func: Awaitable invoked per category.
|
|
proxies: Pool bound to each cycle's fresh session.
|
|
collection: Mongo collection forwarded to every callback.
|
|
headers: Monitor-wide static headers merged into every request.
|
|
session_timeout: Per-request timeout on the cycle's session.
|
|
max_clients: Curl handle pool size. Keep at or above the monitor's
|
|
whole fan-out; see get_default_session for why.
|
|
cycle_timeout: Hard budget in seconds for the whole cycle.
|
|
**kwargs: Forwarded to run_monitor_cycle (pre_task_func,
|
|
concurrent_pre_task_func, and per-task kwargs).
|
|
|
|
Returns:
|
|
The cycle's CycleResult, or None when the budget expired and the
|
|
cycle was abandoned.
|
|
"""
|
|
|
|
async with get_default_session(
|
|
proxies, headers, timeout=session_timeout, max_clients=max_clients
|
|
) as session:
|
|
try:
|
|
async with asyncio.timeout(cycle_timeout):
|
|
return await run_monitor_cycle(
|
|
category_identifiers,
|
|
task_func,
|
|
session=session,
|
|
collection=collection,
|
|
**kwargs,
|
|
)
|
|
except TimeoutError:
|
|
LOGGER.error(
|
|
"cycle exceeded budget, abandoning cycle",
|
|
extra={"timeout": cycle_timeout},
|
|
)
|
|
return None
|
|
|
|
|
|
async def execute_pagination(
|
|
category_identifier: str,
|
|
total_products: int,
|
|
products_per_page: int,
|
|
func: Callable[..., Awaitable[None]],
|
|
session: aiohttp.ClientSession | AsyncSession,
|
|
*,
|
|
semaphore: asyncio.Semaphore | None = None,
|
|
start_offset: int = 0,
|
|
pagination_mode: str = "offset",
|
|
**kwargs,
|
|
):
|
|
"""Paginate a category by dispatching concurrent page or offset tasks to func.
|
|
|
|
Runs pages 2..N under a semaphore, retries failed pages with exponential
|
|
backoff, and logs any pages that still fail after retries. Page 1 is
|
|
assumed to have been fetched by the caller.
|
|
|
|
Args:
|
|
category_identifier: Label used for logging only; not passed to func.
|
|
total_products: Total product count used to derive page count.
|
|
products_per_page: Page size used to derive page count and offsets.
|
|
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.
|
|
semaphore: Optional bound on concurrent page fetches. Omit it to let a
|
|
monitor with few pages fan out unbounded; pass one to cap monitors
|
|
with many pages. Pass a single 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.
|
|
pagination_mode: Either offset or page, selecting the kwarg sent to func.
|
|
**kwargs: Forwarded verbatim to every func call. Any identifier func
|
|
needs (category_id, category_name, category_path, or bestbuy's
|
|
query_category_id) must be supplied here, since naming varies
|
|
per monitor.
|
|
"""
|
|
|
|
total_pages = (int(total_products) + products_per_page - 1) // products_per_page
|
|
LOGGER.info(
|
|
"executing pagination",
|
|
extra={"category": category_identifier, "total": total_pages},
|
|
)
|
|
stop_event = asyncio.Event()
|
|
pagination_label = "offset" if pagination_mode == "offset" else "page"
|
|
page_range = range(2, total_pages + 1)
|
|
|
|
async def fetch_page(page_index: int):
|
|
"""Fetch a single page or offset, optionally under the shared semaphore."""
|
|
async with semaphore or nullcontext():
|
|
if stop_event.is_set():
|
|
return
|
|
|
|
if pagination_mode == "offset":
|
|
current_offset = start_offset + (page_index - 1) * products_per_page
|
|
await func(
|
|
session=session,
|
|
offset=current_offset,
|
|
**kwargs,
|
|
)
|
|
else:
|
|
await func(
|
|
session=session,
|
|
page_number=page_index,
|
|
**kwargs,
|
|
)
|
|
|
|
failed_pages = await fetch_all_pages(
|
|
page_range, fetch_page, category_identifier, pagination_label
|
|
)
|
|
if failed_pages and not stop_event.is_set():
|
|
failed_pages = await retry_failed_pages(
|
|
failed_pages, fetch_page, category_identifier, pagination_label
|
|
)
|
|
|
|
if failed_pages and not stop_event.is_set():
|
|
LOGGER.error(
|
|
"failed to get products after retries",
|
|
extra={
|
|
"category": category_identifier,
|
|
"label": pagination_label,
|
|
"failed": failed_pages,
|
|
},
|
|
)
|
|
|
|
if stop_event.is_set():
|
|
LOGGER.info("pagination stopped early", extra={"category": category_identifier})
|
|
|
|
|
|
async def fetch_all_pages(
|
|
page_range: Iterable[int],
|
|
fetch_func: Callable[[int], Awaitable[None]],
|
|
category_identifier: str,
|
|
pagination_label: str,
|
|
) -> list[int]:
|
|
"""Dispatch all page fetches concurrently and return failed page numbers.
|
|
|
|
Args:
|
|
page_range: Page numbers to fetch.
|
|
fetch_func: Awaitable invoked per page number.
|
|
category_identifier: Label used for logging.
|
|
pagination_label: Either offset or page, used for logging.
|
|
|
|
Returns:
|
|
Page numbers whose fetch raised an expected error.
|
|
"""
|
|
|
|
tasks = [fetch_func(page) for page in page_range]
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
return handle_page_results(
|
|
list(page_range), results, category_identifier, pagination_label
|
|
)
|
|
|
|
|
|
def handle_page_results(
|
|
page_numbers: list[int],
|
|
results: list,
|
|
category_identifier: str,
|
|
pagination_label: str,
|
|
retry_attempt: int | None = None,
|
|
) -> list[int]:
|
|
"""Log page results and collect failures, raising on unexpected errors.
|
|
|
|
Expected errors are logged as warnings and their pages collected for
|
|
retry. Unexpected exceptions are logged and re-raised. Successful pages
|
|
are logged only when part of a retry attempt.
|
|
|
|
Args:
|
|
page_numbers: Page numbers matching results by position.
|
|
results: Gathered results or exceptions per page.
|
|
category_identifier: Label used for logging.
|
|
pagination_label: Either offset or page, used for logging.
|
|
retry_attempt: Retry attempt number when handling retry results.
|
|
|
|
Returns:
|
|
Page numbers whose fetch raised an expected error.
|
|
"""
|
|
|
|
failed_pages = []
|
|
for page_number, result in zip(page_numbers, results):
|
|
if isinstance(result, EXPECTED_TASK_ERRORS):
|
|
LOGGER.warning(
|
|
"failed",
|
|
extra={
|
|
"category": category_identifier,
|
|
"label": pagination_label,
|
|
"page": page_number,
|
|
"retry": retry_attempt,
|
|
},
|
|
exc_info=result,
|
|
)
|
|
failed_pages.append(page_number)
|
|
elif isinstance(result, Exception):
|
|
LOGGER.error(
|
|
"unexpected error",
|
|
extra={
|
|
"category": category_identifier,
|
|
"label": pagination_label,
|
|
"page": page_number,
|
|
"retry": retry_attempt,
|
|
},
|
|
exc_info=result,
|
|
)
|
|
raise result
|
|
elif retry_attempt:
|
|
LOGGER.info(
|
|
"retry succeeded",
|
|
extra={
|
|
"category": category_identifier,
|
|
"label": pagination_label,
|
|
"page": page_number,
|
|
"retry": retry_attempt,
|
|
},
|
|
)
|
|
|
|
return failed_pages
|
|
|
|
|
|
async def retry_failed_pages(
|
|
failed_pages: list[int],
|
|
fetch_func: Callable[[int], Awaitable[None]],
|
|
category_identifier: str,
|
|
pagination_label: str,
|
|
max_retries: int = 3,
|
|
) -> list[int]:
|
|
"""Retry failed pages with exponential backoff until exhausted or clear.
|
|
|
|
Sleeps 2**attempt seconds before each round and re-dispatches only the
|
|
pages that failed the previous round.
|
|
|
|
Args:
|
|
failed_pages: Page numbers to retry.
|
|
fetch_func: Awaitable invoked per page number.
|
|
category_identifier: Label used for logging.
|
|
pagination_label: Either offset or page, used for logging.
|
|
max_retries: Maximum retry rounds.
|
|
|
|
Returns:
|
|
Page numbers still failing after all retries.
|
|
"""
|
|
|
|
retry_attempt = 1
|
|
while failed_pages and retry_attempt <= max_retries:
|
|
await asyncio.sleep(2**retry_attempt)
|
|
LOGGER.info(
|
|
"retrying failed pages",
|
|
extra={
|
|
"category": category_identifier,
|
|
"label": pagination_label,
|
|
"count": len(failed_pages),
|
|
"attempt": retry_attempt,
|
|
"max_retries": max_retries,
|
|
},
|
|
)
|
|
tasks = [fetch_func(page_number) for page_number in failed_pages]
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
failed_pages = handle_page_results(
|
|
failed_pages, results, category_identifier, pagination_label, retry_attempt
|
|
)
|
|
retry_attempt += 1
|
|
|
|
return failed_pages
|
|
|
|
|
|
def first_error(group_exc: BaseExceptionGroup) -> BaseException:
|
|
"""Return a leaf exception of a possibly nested group, preferring an
|
|
unexpected one so a real bug is never masked by a routine network error."""
|
|
_, unexpected = group_exc.split(EXPECTED_TASK_ERRORS)
|
|
if unexpected is not None:
|
|
group_exc = unexpected
|
|
|
|
error = group_exc.exceptions[0]
|
|
while isinstance(error, BaseExceptionGroup):
|
|
error = error.exceptions[0]
|
|
|
|
return error
|
|
|
|
|
|
@asynccontextmanager
|
|
async def fan_out_group():
|
|
"""A TaskGroup for page-one fan-out whose failure surfaces unwrapped.
|
|
|
|
Use it where one page's processing must be bound to the pagination that
|
|
page spawns: create the processing task on the yielded group, then await
|
|
execute_pagination in the body. The group cancels whichever sibling is
|
|
still running when the other fails, so a dead category stops doing work
|
|
nobody will use, and a cancelled cycle takes both down together.
|
|
|
|
The unwrapping is the reason this is shared rather than inlined. A bare
|
|
TaskGroup always raises an ExceptionGroup, and run_monitor_cycle classifies
|
|
task results with isinstance against EXPECTED_TASK_ERRORS -- an
|
|
ExceptionGroup matches neither branch as intended, so a routine page
|
|
failure would be reclassified as an unexpected one and kill the monitor.
|
|
first_error unwraps to a leaf, preferring an unexpected exception so a real
|
|
bug is never masked by a network error.
|
|
|
|
Note the caller's own body is inside the group, so a plain exception raised
|
|
there (not from a child) propagates unchanged rather than being wrapped,
|
|
and an early `return` still awaits the child tasks.
|
|
|
|
Example:
|
|
async with fan_out_group() as group:
|
|
group.create_task(process_page(data, ...))
|
|
await execute_pagination(...)
|
|
"""
|
|
|
|
try:
|
|
async with asyncio.TaskGroup() as group:
|
|
yield group
|
|
except BaseExceptionGroup as group_exc:
|
|
raise first_error(group_exc) from None
|