Run crawls¶

async def run(
    config: str | Path | WebsiteConfig,
    *,
    options: RunOptions | None = None,
    middleware: Sequence[Middleware] = (),
    fetcher: Fetcher | None = None,
) -> RunReport:
    """
    Run one crawl and return its report.

    ``config`` is a config file or a loaded config. ``middleware`` runs after the
    ``[[middleware]]`` entries. ``run`` closes the entries it builds, and the caller closes
    ``middleware``. ``fetcher`` replaces the one ``[fetch]`` describes, and ``run`` closes it when
    it ends. On the main thread, the first SIGINT or SIGTERM stops the crawl: requests in flight
    finish, unfinished URLs return to the queue, and the report records the ``interrupted`` stop.
    A caller with its own handler for that signal, or that ignores it, keeps it, and the signal
    gets its previous handler back when ``run`` ends. A loaded config whose setup
    raises, or whose store cannot write its items when it closes, gets a report with the ``error``
    stop, which is also its ``.last_run.json``.

    Raises:
        ConfigError: The config cannot be loaded.
        BlockingIOError: Another process holds the config's store. Nothing is written to it.
        TypeError: ``fetcher`` is a class, or lacks ``fetch`` or ``close``.
    """
    batch = _Batch()
    with _stopping(batch):
        return await _run(config, options or RunOptions(), list(middleware), fetcher, batch)

async def run_many(
    configs: Sequence[str | Path | WebsiteConfig],
    *,
    parallel: int = 1,
    deadline: float | None = None,
    options: RunOptions | None = None,
) -> list[RunReport]:
    """
    Run ``configs``, up to ``parallel`` at once, and return their reports in order.

    The first SIGINT or SIGTERM, or ``deadline`` seconds after the call, stops every running
    crawl as ``run`` stops on a signal, and no config starts after it. A config that never
    started gets a failed report and writes nothing to its store. Signals follow the rule of
    ``run``. A config that cannot load, or whose run raises, gets a failed report with the
    ``error`` stop, and the others run. A handler with ``RunNameFilter`` labels each log record
    with the name of the config that logged it. With more than one config, ``databrew.progress``
    logs each config's result at INFO when that config ends. Raises ValueError when ``parallel``
    is below 1 or ``deadline`` is not a number.
    """
    if parallel < 1:
        raise ValueError(f"parallel is {parallel}, and it must be at least 1")
    if deadline is not None and math.isnan(deadline):
        raise ValueError("deadline is nan, and it must be a number of seconds")
    batch = _Batch()
    slots = asyncio.Semaphore(parallel)

    async def run_one(config: str | Path | WebsiteConfig) -> RunReport:
        name = config.name if isinstance(config, WebsiteConfig) else Path(config).stem
        # gather() runs each call in a task with its own context.
        _run_name.set(name)
        async with slots:
            try:
                report = await _run(config, options or RunOptions(), [], None, batch)
            except Exception as error:
                one_line = isinstance(error, ONE_LINE_ERRORS)
                logger.error("%s", error, exc_info=not one_line)  # noqa: TRY400
                report = _unfinished(name, Stop.ERROR, str(error))
        # The reports return together after the last config ends, so one that ends earlier logs
        # its result here. A lone config's report returns as soon as it ends.
        if len(configs) > 1:
            _run_name.set(report.name)
            progress_logger.info("Result: %s", format_result(report))
        return report

    with _stopping(batch, deadline):
        return list(await asyncio.gather(*map(run_one, configs)))

async def parse(
    config: str | Path | WebsiteConfig,
    url: str,
    *,
    page_type: Literal["listing", "item"] = "listing",
    text: str | None = None,
    save: str | Path | None = None,
) -> dict[str, Any]:
    """
    Return what a run would extract from one page, writing no crawl state.

    The page takes one fetch, with no retry, through the middleware's ``pre_fetch``,
    ``post_fetch``, and ``post_extract``; ``on_error`` and ``on_challenge`` do not run. With
    ``text`` it starts there, without a fetch or ``pre_fetch``. ``save`` keeps the fetched body in
    that file. The result holds ``url``, ``page_type``, ``status``, ``items``,
    ``pagination_links``, ``item_links``, ``diagnostics``, and ``error``, the message of a failure
    or ``None``.

    Raises:
        ConfigError: The config cannot be loaded or built, or names no ``page_type`` extractor.
        ValueError: ``page_type`` is neither ``listing`` nor ``item``.
    """
    if page_type not in ("listing", "item"):
        raise ValueError(f"page_type is {page_type!r}, and it must be listing or item")
    loaded = config if isinstance(config, WebsiteConfig) else load_config(config)
    components = build_components(loaded)
    try:
        extractor = components.listing_extractor
        if page_type == "item":
            if components.item_extractor is None:
                raise ConfigError(f"{loaded.path}: no [item] to read an item page with")
            extractor = components.item_extractor
        chain = MiddlewareChain(components.middleware)
        ctx = MiddlewareContext(url=url, url_type=page_type)
        if text is not None:
            ctx.response = FetchResult(url=url, status_code=200, text=text)
        else:
            ctx = await _fetch_page(components.fetcher, chain, ctx)
            if save is not None and ctx.response is not None:
                await asyncio.to_thread(Path(save).write_text, ctx.response.text)
        if ctx.response is not None and ctx.error is None:
            ctx = await chain.run("post_fetch", ctx)
        if ctx.error is None:
            task = UrlTask(url=url, url_type=page_type)
            ctx, _ = await extract_page(ctx, extractor, chain, task, key_paths(loaded))
        result = ctx.extract_result
        return {
            "url": url,
            "page_type": page_type,
            "status": ctx.response.status_code if ctx.response else None,
            "items": result.items if result else [],
            "pagination_links": result.pagination_links if result else [],
            "item_links": [asdict(link) for link in result.item_links] if result else [],
            "diagnostics": [asdict(each) for each in result.diagnostics] if result else [],
            "error": ctx.error.message if ctx.error else None,
        }
    finally:
        await close_fetcher(components.fetcher)
        await close_middleware(components.middleware)