_basic_crawler.py

# Inspiration: https://github.com/apify/crawlee/blob/v3.7.3/packages/basic-crawler/src/internals/basic-crawler.ts
from __future__ import annotations

import asyncio
import functools
import logging
import signal
import sys
import tempfile
import threading
import traceback
from asyncio import CancelledError
from collections.abc import AsyncGenerator, Awaitable, Callable, Iterable, Sequence
from contextlib import AsyncExitStack, suppress
from datetime import timedelta
from functools import partial
from io import StringIO
from pathlib import Path
from typing import TYPE_CHECKING, Any, Generic, Literal, ParamSpec, cast
from urllib.parse import ParseResult, urlparse
from weakref import WeakKeyDictionary

from cachetools import LRUCache
from tldextract import TLDExtract
from typing_extensions import NotRequired, TypedDict, TypeVar, Unpack, assert_never
from yarl import URL

from crawlee import EnqueueStrategy, Glob, RequestTransformAction, service_locator
from crawlee._autoscaling import AutoscaledPool, Snapshotter, SystemStatus
from crawlee._log_config import configure_logger, get_configured_log_level, string_to_log_level
from crawlee._request import Request, RequestOptions, RequestState
from crawlee._service_locator import ServiceLocator
from crawlee._types import (
    BasicCrawlingContext,
    EnqueueLinksKwargs,
    ExportDataCsvKwargs,
    ExportDataJsonKwargs,
    GetKeyValueStoreFromRequestHandlerFunction,
    HttpHeaders,
    HttpPayload,
    LogLevel,
    RequestHandlerRunResult,
    SendRequestFunction,
    SkippedReason,
)
from crawlee._utils.docs import docs_group
from crawlee._utils.file import atomic_write, export_csv_to_stream, export_json_to_stream
from crawlee._utils.http import parse_retry_after_header
from crawlee._utils.recurring_task import RecurringTask
from crawlee._utils.robots import RobotsTxtFile
from crawlee._utils.urls import convert_to_absolute_url, is_url_absolute
from crawlee._utils.wait import wait_for
from crawlee._utils.web import is_status_code_client_error, is_status_code_server_error
from crawlee.errors import (
    ContextPipelineInitializationError,
    ContextPipelineInterruptedError,
    HttpClientStatusCodeError,
    HttpStatusCodeError,
    RequestCollisionError,
    RequestHandlerError,
    SessionError,
    UserDefinedErrorHandlerError,
    UserHandlerTimeoutError,
)
from crawlee.events._types import Event, EventCrawlerStatusData
from crawlee.http_clients import ImpitHttpClient
from crawlee.request_loaders import ThrottlingRequestManager
from crawlee.router import Router
from crawlee.sessions import SessionPool
from crawlee.statistics import Statistics, StatisticsState
from crawlee.storages import Dataset, KeyValueStore, RequestQueue

from ._context_pipeline import ContextPipeline
from ._context_utils import swapped_context
from ._logging_utils import (
    get_one_line_error_summary_if_possible,
    reduce_asyncio_timeout_error_to_relevant_traceback_parts,
)

if TYPE_CHECKING:
    import re
    from collections.abc import Iterator
    from contextlib import AbstractAsyncContextManager

    from crawlee._types import (
        ConcurrencySettings,
        EnqueueLinksFunction,
        ExtractLinksFunction,
        GetDataKwargs,
        HttpMethod,
        JsonSerializable,
        PushDataKwargs,
    )
    from crawlee.configuration import Configuration
    from crawlee.events import EventManager
    from crawlee.http_clients import HttpClient, HttpResponse
    from crawlee.proxy_configuration import ProxyConfiguration, ProxyInfo
    from crawlee.request_loaders import RequestManager
    from crawlee.sessions import Session
    from crawlee.statistics import FinalStatistics
    from crawlee.storage_clients import StorageClient
    from crawlee.storage_clients.models import DatasetItemsListPage

TCrawlingContext = TypeVar('TCrawlingContext', bound=BasicCrawlingContext, default=BasicCrawlingContext)
TStatisticsState = TypeVar('TStatisticsState', bound=StatisticsState, default=StatisticsState)
TRequestIterator = TypeVar('TRequestIterator', str, Request)
TParams = ParamSpec('TParams')
T = TypeVar('T')

ErrorHandler = Callable[[TCrawlingContext, Exception], Awaitable[Request | None]]
FailedRequestHandler = Callable[[TCrawlingContext, Exception], Awaitable[None]]
SkippedRequestCallback = Callable[[str, SkippedReason], Awaitable[None]]


class _BasicCrawlerOptions(TypedDict):
    """Non-generic options the `BasicCrawler` constructor."""

    configuration: NotRequired[Configuration]
    """The `Configuration` instance. Some of its properties are used as defaults for the crawler."""

    event_manager: NotRequired[EventManager]
    """The event manager for managing events for the crawler and all its components."""

    storage_client: NotRequired[StorageClient]
    """The storage client for managing storages for the crawler and all its components."""

    request_manager: NotRequired[RequestManager]
    """Manager of requests that should be processed by the crawler."""

    session_pool: NotRequired[SessionPool]
    """A custom `SessionPool` instance, allowing the use of non-default configuration."""

    proxy_configuration: NotRequired[ProxyConfiguration]
    """HTTP proxy configuration used when making requests."""

    http_client: NotRequired[HttpClient]
    """HTTP client used by `BasicCrawlingContext.send_request` method."""

    max_request_retries: NotRequired[int]
    """Specifies the maximum number of retries allowed for a request if its processing fails.
    This includes retries due to navigation errors or errors thrown from user-supplied functions
    (`request_handler`, `pre_navigation_hooks` etc.).

    This limit does not apply to retries triggered by session rotation (see `max_session_rotations`)."""

    max_requests_per_crawl: NotRequired[int | None]
    """Maximum number of pages to open during a crawl. The crawl stops upon reaching this limit.
    Setting this value can help avoid infinite loops in misconfigured crawlers. `None` means no limit.
    Due to concurrency settings, the actual number of pages visited may slightly exceed this value."""

    max_session_rotations: NotRequired[int]
    """Maximum number of session rotations per request. The crawler rotates the session if a proxy error occurs
    or if the website blocks the request.

    The session rotations are not counted towards the `max_request_retries` limit.
    """

    max_crawl_depth: NotRequired[int | None]
    """Specifies the maximum crawl depth. If set, the crawler will stop processing links beyond this depth.
    The crawl depth starts at 0 for initial requests and increases with each subsequent level of links.
    Requests at the maximum depth will still be processed, but no new links will be enqueued from those requests.
    If not set, crawling continues without depth restrictions.
    """

    use_session_pool: NotRequired[bool]
    """Enable the use of a session pool for managing sessions during crawling."""

    retry_on_blocked: NotRequired[bool]
    """If True, the crawler attempts to bypass bot protections automatically."""

    concurrency_settings: NotRequired[ConcurrencySettings]
    """Settings to fine-tune concurrency levels."""

    request_handler_timeout: NotRequired[timedelta]
    """Maximum duration allowed for a single request handler to run."""

    abort_on_error: NotRequired[bool]
    """If True, the crawler stops immediately when any request handler error occurs."""

    configure_logging: NotRequired[bool]
    """If True, the crawler will set up logging infrastructure automatically."""

    statistics_log_format: NotRequired[Literal['table', 'inline']]
    """If 'table', displays crawler statistics as formatted tables in logs. If 'inline', outputs statistics as plain
    text log messages.
    """

    keep_alive: NotRequired[bool]
    """Flag that can keep crawler running even when there are no requests in queue."""

    additional_http_error_status_codes: NotRequired[Iterable[int]]
    """Additional HTTP status codes to treat as errors, triggering automatic retries when encountered."""

    ignore_http_error_status_codes: NotRequired[Iterable[int]]
    """HTTP status codes that are typically considered errors but should be treated as successful responses."""

    _additional_context_managers: NotRequired[Sequence[AbstractAsyncContextManager]]
    """Additional context managers used throughout the crawler lifecycle. Intended for use by
    subclasses rather than direct instantiation of `BasicCrawler`."""

    _logger: NotRequired[logging.Logger]
    """A logger instance, typically provided by a subclass, for consistent logging labels. Intended for use by
    subclasses rather than direct instantiation of `BasicCrawler`."""

    respect_robots_txt_file: NotRequired[bool]
    """If set to `True`, the crawler will automatically try to fetch the robots.txt file for each domain,
    and skip those that are not allowed. This also prevents disallowed URLs to be added via `EnqueueLinksFunction`."""

    status_message_logging_interval: NotRequired[timedelta]
    """Interval for logging the crawler status messages."""

    status_message_callback: NotRequired[
        Callable[[StatisticsState, StatisticsState | None, str], Awaitable[str | None]]
    ]
    """Allows overriding the default status message. The default status message is provided in the parameters.
    Returning `None` suppresses the status message."""

    id: NotRequired[int]
    """Identifier used for crawler state tracking. Use the same id across multiple crawlers to share state between
    them."""


class _BasicCrawlerOptionsGeneric(TypedDict, Generic[TCrawlingContext, TStatisticsState]):
    """Generic options the `BasicCrawler` constructor."""

    request_handler: NotRequired[Callable[[TCrawlingContext], Awaitable[None]]]
    """A callable responsible for handling requests."""

    _context_pipeline: NotRequired[ContextPipeline[TCrawlingContext]]
    """Enables extending the request lifecycle and modifying the crawling context. Intended for use by
    subclasses rather than direct instantiation of `BasicCrawler`."""

    statistics: NotRequired[Statistics[TStatisticsState]]
    """A custom `Statistics` instance, allowing the use of non-default configuration."""


class BasicCrawlerOptions(
    _BasicCrawlerOptions,
    _BasicCrawlerOptionsGeneric[TCrawlingContext, TStatisticsState],
    Generic[TCrawlingContext, TStatisticsState],
):
    """Arguments for the `BasicCrawler` constructor.

    It is intended for typing forwarded `__init__` arguments in the subclasses.
    """


@docs_group('Crawlers')
class BasicCrawler(Generic[TCrawlingContext, TStatisticsState]):
    """A basic web crawler providing a framework for crawling websites.

    The `BasicCrawler` provides a low-level functionality for crawling websites, allowing users to define their
    own page download and data extraction logic. It is designed mostly to be subclassed by crawlers with specific
    purposes. In most cases, you will want to use a more specialized crawler, such as `HttpCrawler`,
    `BeautifulSoupCrawler`, `ParselCrawler`, or `PlaywrightCrawler`. If you are an advanced user and want full
    control over the crawling process, you can subclass the `BasicCrawler` and implement the request-handling logic
    yourself.

    The crawling process begins with URLs provided by a `RequestProvider` instance. Each request is then
    handled by a user-defined `request_handler` function, which processes the page and extracts the data.

    The `BasicCrawler` includes several common features for crawling, such as:
        - automatic scaling based on the system resources,
        - retries for failed requests,
        - session management,
        - statistics tracking,
        - request routing via labels,
        - proxy rotation,
        - direct storage interaction helpers,
        - and more.
    """

    _CRAWLEE_STATE_KEY = 'CRAWLEE_STATE'
    _request_handler_timeout_text = 'Request handler timed out after'
    __next_id = 0

    def __init__(
        self,
        *,
        configuration: Configuration | None = None,
        event_manager: EventManager | None = None,
        storage_client: StorageClient | None = None,
        request_manager: RequestManager | None = None,
        session_pool: SessionPool | None = None,
        proxy_configuration: ProxyConfiguration | None = None,
        http_client: HttpClient | None = None,
        request_handler: Callable[[TCrawlingContext], Awaitable[None]] | None = None,
        max_request_retries: int = 3,
        max_requests_per_crawl: int | None = None,
        max_session_rotations: int = 10,
        max_crawl_depth: int | None = None,
        use_session_pool: bool = True,
        retry_on_blocked: bool = True,
        additional_http_error_status_codes: Iterable[int] | None = None,
        ignore_http_error_status_codes: Iterable[int] | None = None,
        concurrency_settings: ConcurrencySettings | None = None,
        request_handler_timeout: timedelta = timedelta(minutes=1),
        statistics: Statistics[TStatisticsState] | None = None,
        abort_on_error: bool = False,
        keep_alive: bool = False,
        configure_logging: bool = True,
        statistics_log_format: Literal['table', 'inline'] = 'table',
        respect_robots_txt_file: bool = False,
        status_message_logging_interval: timedelta = timedelta(seconds=10),
        status_message_callback: Callable[[StatisticsState, StatisticsState | None, str], Awaitable[str | None]]
        | None = None,
        id: int | None = None,
        _context_pipeline: ContextPipeline[TCrawlingContext] | None = None,
        _additional_context_managers: Sequence[AbstractAsyncContextManager] | None = None,
        _logger: logging.Logger | None = None,
    ) -> None:
        """Initialize a new instance.

        Args:
            configuration: The `Configuration` instance. Some of its properties are used as defaults for the crawler.
            event_manager: The event manager for managing events for the crawler and all its components.
            storage_client: The storage client for managing storages for the crawler and all its components.
            request_manager: Manager of requests that should be processed by the crawler.
            session_pool: A custom `SessionPool` instance, allowing the use of non-default configuration.
            proxy_configuration: HTTP proxy configuration used when making requests.
            http_client: HTTP client used by `BasicCrawlingContext.send_request` method.
            request_handler: A callable responsible for handling requests.
            max_request_retries: Specifies the maximum number of retries allowed for a request if its processing fails.
                This includes retries due to navigation errors or errors thrown from user-supplied functions
                (`request_handler`, `pre_navigation_hooks` etc.).
                This limit does not apply to retries triggered by session rotation (see `max_session_rotations`).
            max_requests_per_crawl: Maximum number of pages to open during a crawl. The crawl stops upon reaching
                this limit. Setting this value can help avoid infinite loops in misconfigured crawlers. `None` means
                no limit. Due to concurrency settings, the actual number of pages visited may slightly exceed
                this value. If used together with `keep_alive`, then the crawler will be kept alive only until
                `max_requests_per_crawl` is achieved.
            max_session_rotations: Maximum number of session rotations per request. The crawler rotates the session
                if a proxy error occurs or if the website blocks the request.
                The session rotations are not counted towards the `max_request_retries` limit.
            max_crawl_depth: Specifies the maximum crawl depth. If set, the crawler will stop processing links beyond
                this depth. The crawl depth starts at 0 for initial requests and increases with each subsequent level
                of links. Requests at the maximum depth will still be processed, but no new links will be enqueued
                from those requests. If not set, crawling continues without depth restrictions.
            use_session_pool: Enable the use of a session pool for managing sessions during crawling.
            retry_on_blocked: If True, the crawler attempts to bypass bot protections automatically.
            additional_http_error_status_codes: Additional HTTP status codes to treat as errors,
                triggering automatic retries when encountered.
            ignore_http_error_status_codes: HTTP status codes that are typically considered errors but should be treated
                as successful responses.
            concurrency_settings: Settings to fine-tune concurrency levels.
            request_handler_timeout: Maximum duration allowed for a single request handler to run.
            statistics: A custom `Statistics` instance, allowing the use of non-default configuration.
            abort_on_error: If True, the crawler stops immediately when any request handler error occurs.
            keep_alive: If True, it will keep crawler alive even if there are no requests in queue.
                Use `crawler.stop()` to exit the crawler.
            configure_logging: If True, the crawler will set up logging infrastructure automatically.
            statistics_log_format: If 'table', displays crawler statistics as formatted tables in logs. If 'inline',
                outputs statistics as plain text log messages.
            respect_robots_txt_file: If set to `True`, the crawler will automatically try to fetch the robots.txt file
                for each domain, and skip those that are not allowed. This also prevents disallowed URLs to be added
                via `EnqueueLinksFunction`
            status_message_logging_interval: Interval for logging the crawler status messages.
            status_message_callback: Allows overriding the default status message. The default status message is
                provided in the parameters. Returning `None` suppresses the status message.
            id: Identifier used for crawler state tracking. Use the same id across multiple crawlers to share state
                between them.
            _context_pipeline: Enables extending the request lifecycle and modifying the crawling context.
                Intended for use by subclasses rather than direct instantiation of `BasicCrawler`.
            _additional_context_managers: Additional context managers used throughout the crawler lifecycle.
                Intended for use by subclasses rather than direct instantiation of `BasicCrawler`.
            _logger: A logger instance, typically provided by a subclass, for consistent logging labels.
                Intended for use by subclasses rather than direct instantiation of `BasicCrawler`.
        """
        if id is None:
            self._id = BasicCrawler.__next_id
            BasicCrawler.__next_id += 1
        else:
            self._id = id

        implicit_event_manager_with_explicit_config = False
        if not configuration:
            configuration = service_locator.get_configuration()
        elif not event_manager:
            implicit_event_manager_with_explicit_config = True

        if not storage_client:
            storage_client = service_locator.get_storage_client()

        if not event_manager:
            event_manager = service_locator.get_event_manager()

        self._service_locator = ServiceLocator(
            configuration=configuration, storage_client=storage_client, event_manager=event_manager
        )

        config = self._service_locator.get_configuration()

        # Core components
        self._request_manager = request_manager
        self._session_pool = session_pool or SessionPool()
        self._proxy_configuration = proxy_configuration

        self._additional_http_error_status_codes = (
            set(additional_http_error_status_codes) if additional_http_error_status_codes else set()
        )
        self._ignore_http_error_status_codes = (
            set(ignore_http_error_status_codes) if ignore_http_error_status_codes else set()
        )

        self._http_client = http_client or ImpitHttpClient()

        # Request router setup
        self._router: Router[TCrawlingContext] | None = None
        if isinstance(cast('Router', request_handler), Router):
            self._router = cast('Router[TCrawlingContext]', request_handler)
        elif request_handler is not None:
            self._router = None
            self.router.default_handler(request_handler)

        # Error, failed & skipped request handlers
        self._error_handler: ErrorHandler[TCrawlingContext | BasicCrawlingContext] | None = None
        self._failed_request_handler: FailedRequestHandler[TCrawlingContext | BasicCrawlingContext] | None = None
        self._on_skipped_request: SkippedRequestCallback | None = None
        self._abort_on_error = abort_on_error

        # Crawler callbacks
        self._status_message_callback = status_message_callback

        # Context of each request with matching result of request handler.
        # Inheritors can use this to override the result of individual request handler runs in `_run_request_handler`.
        self._context_result_map = WeakKeyDictionary[BasicCrawlingContext, RequestHandlerRunResult]()

        # Context pipeline
        self._context_pipeline = (_context_pipeline or ContextPipeline()).compose(self._check_url_after_redirects)  # ty: ignore[invalid-argument-type]

        # Crawl settings
        self._max_request_retries = max_request_retries
        self._max_requests_per_crawl = max_requests_per_crawl
        self._max_session_rotations = max_session_rotations
        self._max_crawl_depth = max_crawl_depth
        self._respect_robots_txt_file = respect_robots_txt_file

        # Timeouts
        self._request_handler_timeout = request_handler_timeout
        self._internal_timeout = (
            config.internal_timeout
            if config.internal_timeout is not None
            else max(2 * request_handler_timeout, timedelta(minutes=5))
        )

        # Retry and session settings
        self._use_session_pool = use_session_pool
        self._retry_on_blocked = retry_on_blocked

        # Logging setup
        if configure_logging:
            root_logger = logging.getLogger()
            configure_logger(root_logger, remove_old_handlers=True)
            httpx_logger = logging.getLogger('httpx')  # Silence HTTPX logger
            httpx_logger.setLevel(logging.DEBUG if get_configured_log_level() <= logging.DEBUG else logging.WARNING)
        self._logger = _logger or logging.getLogger(__name__)
        if implicit_event_manager_with_explicit_config:
            self._logger.warning(
                'No event manager set, implicitly using event manager from global service_locator.'
                'It is advised to explicitly set the event manager if explicit configuration is used as well.'
            )
        self._statistics_log_format = statistics_log_format

        # Statistics
        if statistics:
            self._statistics = statistics
        else:

            async def persist_state_factory() -> KeyValueStore:
                return await self.get_key_value_store()

            self._statistics = cast(
                'Statistics[TStatisticsState]',
                Statistics.with_default_state(
                    persistence_enabled=True,
                    periodic_message_logger=self._logger,
                    statistics_log_format=self._statistics_log_format,
                    log_message='Current request statistics:',
                    persist_state_kvs_factory=persist_state_factory,
                ),
            )

        # Additional context managers to enter and exit
        self._additional_context_managers = _additional_context_managers or []

        # Internal, not explicitly configurable components
        self._robots_txt_file_cache: LRUCache[str, RobotsTxtFile] = LRUCache(maxsize=1000)
        self._robots_txt_lock = asyncio.Lock()
        self._tld_extractor = TLDExtract(cache_dir=tempfile.TemporaryDirectory().name)
        self._snapshotter = Snapshotter.from_config(config)
        self._autoscaled_pool = AutoscaledPool(
            system_status=SystemStatus(self._snapshotter),
            concurrency_settings=concurrency_settings,
            is_finished_function=self.__is_finished_function,
            is_task_ready_function=self.__is_task_ready_function,
            run_task_function=self.__run_task_function,
        )
        self._crawler_state_rec_task = RecurringTask(
            func=self._crawler_state_task, delay=status_message_logging_interval
        )
        self._previous_crawler_state: TStatisticsState | None = None

        # State flags
        self._keep_alive = keep_alive
        self._running = False
        self._has_finished_before = False

        self._failed = False

        self._unexpected_stop = False

    @property
    def log(self) -> logging.Logger:
        """The logger used by the crawler."""
        return self._logger

    @property
    def router(self) -> Router[TCrawlingContext]:
        """The `Router` used to handle each individual crawling request."""
        if self._router is None:
            self._router = Router[TCrawlingContext]()

        return self._router

    @router.setter
    def router(self, router: Router[TCrawlingContext]) -> None:
        if self._router is not None:
            raise RuntimeError('A router is already set')

        self._router = router

    @property
    def statistics(self) -> Statistics[TStatisticsState]:
        """Statistics about the current (or last) crawler run."""
        return self._statistics

    def stop(self, reason: str = 'Stop was called externally.') -> None:
        """Set flag to stop crawler.

        This stops current crawler run regardless of whether all requests were finished.

        Args:
            reason: Reason for stopping that will be used in logs.
        """
        self._logger.info(f'Crawler.stop() was called with following reason: {reason}.')
        self._unexpected_stop = True

    def _wrap_handler_with_error_context(
        self, handler: Callable[[TCrawlingContext | BasicCrawlingContext, Exception], Awaitable[T]]
    ) -> Callable[[TCrawlingContext | BasicCrawlingContext, Exception], Awaitable[T]]:
        """Decorate error handlers to make their context helpers usable."""

        @functools.wraps(handler)
        async def wrapped_handler(context: TCrawlingContext | BasicCrawlingContext, exception: Exception) -> T:
            # Original context helpers that are from `RequestHandlerRunResult` will not be committed as the request
            # failed. Modified context provides context helpers with direct access to the storages.
            error_context = context.create_modified_copy(
                push_data=self._push_data,
                get_key_value_store=self.get_key_value_store,
                add_requests=functools.partial(self._add_requests, context),
            )
            return await handler(error_context, exception)

        return wrapped_handler

    def _stop_if_max_requests_count_exceeded(self) -> None:
        """Call `stop` when the maximum number of requests to crawl has been reached."""
        if self._max_requests_per_crawl is None:
            return

        if self._statistics.state.requests_finished >= self._max_requests_per_crawl:
            self.stop(
                reason=f'The crawler has reached its limit of {self._max_requests_per_crawl} requests per crawl. '
            )

    async def _get_session(self) -> Session | None:
        """If session pool is being used, try to take a session from it."""
        if not self._use_session_pool:
            return None

        return await wait_for(
            self._session_pool.get_session,
            timeout=self._internal_timeout,
            timeout_message='Fetching a session from the pool timed out after '
            f'{self._internal_timeout.total_seconds()} seconds',
            max_retries=3,
            logger=self._logger,
        )

    async def _get_session_by_id(self, session_id: str | None) -> Session | None:
        """If session pool is being used, try to take a session by id from it."""
        if not self._use_session_pool or not session_id:
            return None

        return await wait_for(
            partial(self._session_pool.get_session_by_id, session_id),
            timeout=self._internal_timeout,
            timeout_message='Fetching a session from the pool timed out after '
            f'{self._internal_timeout.total_seconds()} seconds',
            max_retries=3,
            logger=self._logger,
        )

    async def _get_proxy_info(self, request: Request, session: Session | None) -> ProxyInfo | None:
        """Retrieve a new ProxyInfo object based on crawler configuration and the current request and session."""
        if not self._proxy_configuration:
            return None

        return await self._proxy_configuration.new_proxy_info(
            session_id=session.id if session else None,
            request=request,
            proxy_tier=None,
        )

    async def get_request_manager(self) -> RequestManager:
        """Return the configured request manager. If none is configured, open and return the default request queue.

        The returned manager is wrapped with `ThrottlingRequestManager` to enforce
        per-domain delays from 429 responses and robots.txt crawl-delay directives.
        """
        if not self._request_manager:
            inner = await RequestQueue.open(
                storage_client=self._service_locator.get_storage_client(),
                configuration=self._service_locator.get_configuration(),
            )
            self._request_manager = ThrottlingRequestManager(inner)

        return self._request_manager

    async def get_dataset(
        self,
        *,
        id: str | None = None,
        name: str | None = None,
        alias: str | None = None,
    ) -> Dataset:
        """Return the `Dataset` with the given ID or name. If none is provided, return the default one."""
        return await Dataset.open(
            id=id,
            name=name,
            alias=alias,
            storage_client=self._service_locator.get_storage_client(),
            configuration=self._service_locator.get_configuration(),
        )

    async def get_key_value_store(
        self,
        *,
        id: str | None = None,
        name: str | None = None,
        alias: str | None = None,
    ) -> KeyValueStore:
        """Return the `KeyValueStore` with the given ID or name. If none is provided, return the default KVS."""
        return await KeyValueStore.open(
            id=id,
            name=name,
            alias=alias,
            storage_client=self._service_locator.get_storage_client(),
            configuration=self._service_locator.get_configuration(),
        )

    def error_handler(
        self, handler: ErrorHandler[TCrawlingContext | BasicCrawlingContext]
    ) -> ErrorHandler[TCrawlingContext]:
        """Register a function to handle errors occurring in request handlers.

        The error handler is invoked after a request handler error occurs and before a retry attempt.
        """
        self._error_handler = self._wrap_handler_with_error_context(handler)
        return handler

    def failed_request_handler(
        self, handler: FailedRequestHandler[TCrawlingContext | BasicCrawlingContext]
    ) -> FailedRequestHandler[TCrawlingContext]:
        """Register a function to handle requests that exceed the maximum retry limit.

        The failed request handler is invoked when a request has failed all retry attempts.
        """
        self._failed_request_handler = self._wrap_handler_with_error_context(handler)
        return handler

    def on_skipped_request(self, callback: SkippedRequestCallback) -> SkippedRequestCallback:
        """Register a function to handle skipped requests.

        The skipped request handler is invoked when a request is skipped due to a collision or other reasons.
        """
        self._on_skipped_request = callback
        return callback

    async def run(
        self,
        requests: Sequence[str | Request] | None = None,
        *,
        purge_request_queue: bool = True,
    ) -> FinalStatistics:
        """Run the crawler until all requests are processed.

        Args:
            requests: The requests to be enqueued before the crawler starts.
            purge_request_queue: If this is `True` and the crawler is not being run for the first time, the default
                request queue will be purged.
        """
        if self._running:
            raise RuntimeError(
                'This crawler instance is already running, you can add more requests to it via `crawler.add_requests()`'
            )

        self._running = True

        if self._has_finished_before:
            await self._statistics.reset()

            if self._use_session_pool:
                await self._session_pool.reset_store()

            request_manager = await self.get_request_manager()
            if purge_request_queue:
                if isinstance(request_manager, RequestQueue):
                    await request_manager.drop()
                    self._request_manager = await RequestQueue.open(
                        storage_client=self._service_locator.get_storage_client(),
                        configuration=self._service_locator.get_configuration(),
                    )
                elif isinstance(request_manager, ThrottlingRequestManager):
                    await request_manager.drop()
                    inner = await RequestQueue.open(
                        storage_client=self._service_locator.get_storage_client(),
                        configuration=self._service_locator.get_configuration(),
                    )
                    self._throttling_manager = ThrottlingRequestManager(inner)
                    self._request_manager = self._throttling_manager

        if requests is not None:
            await self.add_requests(requests)

        interrupted = False

        def sigint_handler() -> None:
            nonlocal interrupted

            if not interrupted:
                interrupted = True
                self._logger.info('Pausing... Press CTRL+C again to force exit.')

            run_task.cancel()

        run_task = asyncio.create_task(self._run_crawler(), name='run_crawler_task')

        if threading.current_thread() is threading.main_thread():  # `add_signal_handler` works only in the main thread
            with suppress(NotImplementedError):  # event loop signal handlers are not supported on Windows
                asyncio.get_running_loop().add_signal_handler(signal.SIGINT, sigint_handler)

        try:
            await run_task
        except CancelledError:
            pass
        finally:
            if threading.current_thread() is threading.main_thread():
                with suppress(NotImplementedError):
                    asyncio.get_running_loop().remove_signal_handler(signal.SIGINT)

        if self._statistics.error_tracker.total > 0:
            self._logger.info(
                'Error analysis:'
                f' total_errors={self._statistics.error_tracker.total}'
                f' unique_errors={self._statistics.error_tracker.unique_error_count}'
            )

        if interrupted:
            self._logger.info(
                f'The crawl was interrupted. To resume, do: CRAWLEE_PURGE_ON_START=0 python {sys.argv[0]}'
            )

        self._running = False
        self._has_finished_before = True

        await self._save_crawler_state()

        final_statistics = self._statistics.calculate()
        if self._statistics_log_format == 'table':
            self._logger.info(f'Final request statistics:\n{final_statistics.to_table()}')
        else:
            self._logger.info('Final request statistics:', extra=final_statistics.to_dict())
        return final_statistics

    async def _run_crawler(self) -> None:
        event_manager = self._service_locator.get_event_manager()

        # Collect the context managers to be entered. Context managers that are already active are excluded,
        # as they were likely entered by the caller, who will also be responsible for exiting them.
        contexts_to_enter = [
            cm
            for cm in (
                event_manager,
                self._snapshotter,
                self._statistics,
                self._session_pool if self._use_session_pool else None,
                self._http_client,
                self._crawler_state_rec_task,
                *self._additional_context_managers,
            )
            if cm and getattr(cm, 'active', False) is False
        ]

        async with AsyncExitStack() as exit_stack:
            for context in contexts_to_enter:
                await exit_stack.enter_async_context(context)  # ty: ignore[invalid-argument-type]

            await self._autoscaled_pool.run()

    async def add_requests(
        self,
        requests: Sequence[str | Request],
        *,
        forefront: bool = False,
        batch_size: int = 1000,
        wait_time_between_batches: timedelta = timedelta(0),
        wait_for_all_requests_to_be_added: bool = False,
        wait_for_all_requests_to_be_added_timeout: timedelta | None = None,
    ) -> None:
        """Add requests to the underlying request manager in batches.

        Args:
            requests: A list of requests to add to the queue.
            forefront: If True, add requests to the forefront of the queue.
            batch_size: The number of requests to add in one batch.
            wait_time_between_batches: Time to wait between adding batches.
            wait_for_all_requests_to_be_added: If True, wait for all requests to be added before returning.
            wait_for_all_requests_to_be_added_timeout: Timeout for waiting for all requests to be added.
        """
        allowed_requests = []
        skipped = []

        for request in requests:
            check_url = request.url if isinstance(request, Request) else request
            if await self._is_allowed_based_on_robots_txt_file(check_url):
                allowed_requests.append(request)
            else:
                skipped.append(request)

        if skipped:
            skipped_tasks = [
                asyncio.create_task(self._handle_skipped_request(request, 'robots_txt')) for request in skipped
            ]
            await asyncio.gather(*skipped_tasks)
            self._logger.warning('Some requests were skipped because they were disallowed based on the robots.txt file')

        request_manager = await self.get_request_manager()

        await request_manager.add_requests(
            requests=allowed_requests,
            forefront=forefront,
            batch_size=batch_size,
            wait_time_between_batches=wait_time_between_batches,
            wait_for_all_requests_to_be_added=wait_for_all_requests_to_be_added,
            wait_for_all_requests_to_be_added_timeout=wait_for_all_requests_to_be_added_timeout,
        )

    async def _use_state(
        self,
        default_value: dict[str, JsonSerializable] | None = None,
    ) -> dict[str, JsonSerializable]:
        kvs = await self.get_key_value_store()
        return await kvs.get_auto_saved_value(f'{self._CRAWLEE_STATE_KEY}_{self._id}', default_value)

    async def _save_crawler_state(self) -> None:
        store = await self.get_key_value_store()
        await store.persist_autosaved_values()

    async def get_data(
        self,
        dataset_id: str | None = None,
        dataset_name: str | None = None,
        dataset_alias: str | None = None,
        **kwargs: Unpack[GetDataKwargs],
    ) -> DatasetItemsListPage:
        """Retrieve data from a `Dataset`.

        This helper method simplifies the process of retrieving data from a `Dataset`. It opens the specified
        one and then retrieves the data based on the provided parameters.

        Args:
            dataset_id: The ID of the `Dataset`.
            dataset_name: The name of the `Dataset` (global scope, named storage).
            dataset_alias: The alias of the `Dataset` (run scope, unnamed storage).
            kwargs: Keyword arguments to be passed to the `Dataset.get_data()` method.

        Returns:
            The retrieved data.
        """
        dataset = await Dataset.open(
            id=dataset_id,
            name=dataset_name,
            alias=dataset_alias,
            storage_client=self._service_locator.get_storage_client(),
            configuration=self._service_locator.get_configuration(),
        )
        return await dataset.get_data(**kwargs)

    async def export_data(
        self,
        path: str | Path,
        dataset_id: str | None = None,
        dataset_name: str | None = None,
        dataset_alias: str | None = None,
        **additional_kwargs: Unpack[ExportDataJsonKwargs | ExportDataCsvKwargs],
    ) -> None:
        """Export all items from a Dataset to a JSON or CSV file.

        This method simplifies the process of exporting data collected during crawling. It automatically
        determines the export format based on the file extension (`.json` or `.csv`) and handles
        the conversion of `Dataset` items to the appropriate format.

        Args:
            path: The destination file path. Must end with '.json' or '.csv'.
            dataset_id: The ID of the Dataset to export from.
            dataset_name: The name of the Dataset to export from (global scope, named storage).
            dataset_alias: The alias of the Dataset to export from (run scope, unnamed storage).
            additional_kwargs: Extra keyword arguments forwarded to the JSON/CSV exporter depending on the file format.
        """
        dataset = await Dataset.open(
            id=dataset_id,
            name=dataset_name,
            alias=dataset_alias,
            storage_client=self._service_locator.get_storage_client(),
            configuration=self._service_locator.get_configuration(),
        )

        path = Path(path)

        if path.suffix == '.csv':
            dst = StringIO()
            csv_kwargs = cast('ExportDataCsvKwargs', additional_kwargs)
            await export_csv_to_stream(dataset.iterate_items(), dst, **csv_kwargs)
            await atomic_write(path, dst.getvalue())
        elif path.suffix == '.json':
            dst = StringIO()
            json_kwargs = cast('ExportDataJsonKwargs', additional_kwargs)
            await export_json_to_stream(dataset.iterate_items(), dst, **json_kwargs)
            await atomic_write(path, dst.getvalue())
        else:
            raise ValueError(f'Unsupported file extension: {path.suffix}')

    async def _push_data(
        self,
        data: list[dict[str, Any]] | dict[str, Any],
        dataset_id: str | None = None,
        dataset_name: str | None = None,
        dataset_alias: str | None = None,
        **kwargs: Unpack[PushDataKwargs],
    ) -> None:
        """Push data to a `Dataset`.

        This helper method simplifies the process of pushing data to a `Dataset`. It opens the specified
        one and then pushes the provided data to it.

        Args:
            data: The data to push to the `Dataset`.
            dataset_id: The ID of the `Dataset`.
            dataset_name: The name of the `Dataset` (global scope, named storage).
            dataset_alias: The alias of the `Dataset` (run scope, unnamed storage).
            kwargs: Keyword arguments to be passed to the `Dataset.push_data()` method.
        """
        dataset = await self.get_dataset(id=dataset_id, name=dataset_name, alias=dataset_alias)
        await dataset.push_data(data, **kwargs)

    def _should_retry_request(self, context: BasicCrawlingContext, error: Exception) -> bool:
        if context.request.no_retry:
            return False

        # Do not retry on client errors.
        if isinstance(error, HttpClientStatusCodeError):
            return False

        if isinstance(error, SessionError):
            return ((context.request.session_rotation_count or 0) + 1) < self._max_session_rotations

        max_request_retries = context.request.max_retries
        if max_request_retries is None:
            max_request_retries = self._max_request_retries

        return context.request.retry_count < max_request_retries

    async def _check_url_after_redirects(self, context: TCrawlingContext) -> AsyncGenerator[TCrawlingContext, None]:
        """Ensure that the `loaded_url` still matches the enqueue strategy after redirects.

        Filter out links that redirect outside of the crawled domain.
        """
        if context.request.loaded_url is not None and not self._check_enqueue_strategy(
            context.request.enqueue_strategy,
            origin_url=urlparse(context.request.url),
            target_url=urlparse(context.request.loaded_url),
        ):
            raise ContextPipelineInterruptedError(
                f'Skipping URL {context.request.loaded_url} (redirected from {context.request.url})'
            )

        yield context

    def _create_enqueue_links_function(
        self, context: BasicCrawlingContext, extract_links: ExtractLinksFunction
    ) -> EnqueueLinksFunction:
        """Create a callback function for extracting links from parsed content and enqueuing them to the crawl.

        Args:
            context: The current crawling context.
            extract_links: Function used to extract links from the page.

        Returns:
            Awaitable that is used for extracting links from parsed content and enqueuing them to the crawl.
        """

        async def enqueue_links(
            *,
            selector: str | None = None,
            label: str | None = None,
            user_data: dict[str, Any] | None = None,
            transform_request_function: Callable[[RequestOptions], RequestOptions | RequestTransformAction]
            | None = None,
            requests: Sequence[str | Request] | None = None,
            rq_id: str | None = None,
            rq_name: str | None = None,
            rq_alias: str | None = None,
            **kwargs: Unpack[EnqueueLinksKwargs],
        ) -> None:
            kwargs.setdefault('strategy', 'same-hostname')

            if requests:
                if any((selector, label, user_data, transform_request_function)):
                    raise ValueError(
                        'You cannot provide `selector`, `label`, `user_data` or '
                        '`transform_request_function` arguments when `requests` is provided.'
                    )
                # Add directly passed requests.
                await context.add_requests(
                    requests or list[str | Request](), rq_id=rq_id, rq_name=rq_name, rq_alias=rq_alias, **kwargs
                )
            else:
                # Add requests from extracted links.
                await context.add_requests(
                    await extract_links(
                        selector=selector or 'a',
                        label=label,
                        user_data=user_data,
                        transform_request_function=transform_request_function,
                        **kwargs,
                    ),
                    rq_id=rq_id,
                    rq_name=rq_name,
                    rq_alias=rq_alias,
                    **kwargs,
                )

        return enqueue_links

    def _enqueue_links_filter_iterator(
        self, request_iterator: Iterator[TRequestIterator], origin_url: str, **kwargs: Unpack[EnqueueLinksKwargs]
    ) -> Iterator[TRequestIterator]:
        """Filter requests based on the enqueue strategy and URL patterns."""
        limit = kwargs.get('limit')
        parsed_origin_url = urlparse(origin_url)
        strategy = kwargs.get('strategy', 'all')

        if strategy == 'all' and not parsed_origin_url.hostname:
            self.log.warning(f'Skipping enqueue: Missing hostname in origin_url = {origin_url}.')
            return

        # Emit a `warning` message to the log, only once per call
        warning_flag = True

        for request in request_iterator:
            if isinstance(request, Request):
                if request.enqueue_strategy != strategy:
                    request.enqueue_strategy = strategy
                target_url = request.url
            else:
                target_url = request
            parsed_target_url = urlparse(target_url)

            if warning_flag and strategy != 'all' and not parsed_target_url.hostname:
                self.log.warning(f'Skipping enqueue url: Missing hostname in target_url = {target_url}.')
                warning_flag = False

            if self._check_enqueue_strategy(
                strategy, target_url=parsed_target_url, origin_url=parsed_origin_url
            ) and self._check_url_patterns(target_url, kwargs.get('include'), kwargs.get('exclude')):
                yield request

                if limit is not None:
                    limit -= 1
                    if limit <= 0:
                        break

    def _check_enqueue_strategy(
        self,
        strategy: EnqueueStrategy,
        *,
        target_url: ParseResult,
        origin_url: ParseResult,
    ) -> bool:
        """Check if a URL matches the enqueue_strategy."""
        if strategy == 'all':
            return True

        if origin_url.hostname is None or target_url.hostname is None:
            self.log.debug(
                f'Skipping enqueue: Missing hostname in origin_url = {origin_url.geturl()} or '
                f'target_url = {target_url.geturl()}'
            )
            return False

        if strategy == 'same-hostname':
            return target_url.hostname == origin_url.hostname

        if strategy == 'same-domain':
            origin_domain = self._tld_extractor.extract_str(origin_url.hostname).top_domain_under_public_suffix
            target_domain = self._tld_extractor.extract_str(target_url.hostname).top_domain_under_public_suffix
            return origin_domain == target_domain

        if strategy == 'same-origin':
            return (
                target_url.hostname == origin_url.hostname
                and target_url.scheme == origin_url.scheme
                and target_url.port == origin_url.port
            )

        assert_never(strategy)

    def _check_url_patterns(
        self,
        target_url: str,
        include: Sequence[re.Pattern[Any] | Glob] | None,
        exclude: Sequence[re.Pattern[Any] | Glob] | None,
    ) -> bool:
        """Check if a URL matches configured include/exclude patterns."""
        # If the URL matches any `exclude` pattern, reject it
        for pattern in exclude or ():
            if isinstance(pattern, Glob):
                pattern = pattern.regexp  # noqa: PLW2901

            if pattern.match(target_url) is not None:
                return False

        # If there are no `include` patterns and the URL passed all `exclude` patterns, accept the URL
        if include is None:
            return True

        # If the URL matches any `include` pattern, accept it
        for pattern in include:
            if isinstance(pattern, Glob):
                pattern = pattern.regexp  # noqa: PLW2901

            if pattern.match(target_url) is not None:
                return True

        # The URL does not match any `include` pattern - reject it
        return False

    async def _handle_request_retries(
        self,
        context: TCrawlingContext | BasicCrawlingContext,
        error: Exception,
    ) -> None:
        request_manager = await self.get_request_manager()
        request = context.request

        if self._abort_on_error:
            self._logger.exception('Aborting crawler run due to error (abort_on_error=True)', exc_info=error)
            self._failed = True

        if self._should_retry_request(context, error):
            request.retry_count += 1
            reduced_error = str(error).split('\n')[0]
            self.log.warning(
                f'Retrying request to {context.request.url} due to: {reduced_error}. '
                f'{get_one_line_error_summary_if_possible(error)}'
            )
            await self._statistics.error_tracker.add(error=error, context=context)

            if self._error_handler:
                try:
                    new_request = await self._error_handler(context, error)
                except Exception as e:
                    raise UserDefinedErrorHandlerError('Exception thrown in user-defined request error handler') from e
                else:
                    if new_request is not None and new_request != request:
                        await request_manager.add_request(new_request)
                        await self._mark_request_as_handled(request)
                        return

            await request_manager.reclaim_request(request)
        else:
            request.state = RequestState.ERROR
            await self._mark_request_as_handled(request)
            await self._handle_failed_request(context, error)
            self._statistics.record_request_processing_failure(request.unique_key)

    async def _handle_request_error(self, context: TCrawlingContext | BasicCrawlingContext, error: Exception) -> None:
        try:
            context.request.state = RequestState.ERROR_HANDLER

            await wait_for(
                partial(self._handle_request_retries, context, error),
                timeout=self._internal_timeout,
                timeout_message='Handling request failure timed out after '
                f'{self._internal_timeout.total_seconds()} seconds',
                logger=self._logger,
            )
        except UserDefinedErrorHandlerError:
            context.request.state = RequestState.ERROR
            raise
        except Exception as secondary_error:
            self._logger.exception(
                'An exception occurred during handling of failed request. This places the crawler '
                'and its underlying storages into an unknown state and crawling will be terminated.',
                exc_info=secondary_error,
            )
            context.request.state = RequestState.ERROR
            raise

        if context.session:
            context.session.mark_bad()

    async def _handle_failed_request(self, context: TCrawlingContext | BasicCrawlingContext, error: Exception) -> None:
        self._logger.error(
            f'Request to {context.request.url} failed and reached maximum retries\n '
            f'{self._get_message_from_error(error)}'
        )
        await self._statistics.error_tracker.add(error=error, context=context)

        if self._failed_request_handler:
            try:
                await self._failed_request_handler(context, error)
            except Exception as e:
                raise UserDefinedErrorHandlerError('Exception thrown in user-defined failed request handler') from e

    async def _handle_skipped_request(
        self, request: Request | str, reason: SkippedReason, *, need_mark: bool = False
    ) -> None:
        if need_mark and isinstance(request, Request):
            request.state = RequestState.SKIPPED
            await self._mark_request_as_handled(request)

        url = request.url if isinstance(request, Request) else request

        if self._on_skipped_request:
            try:
                await self._on_skipped_request(url, reason)
            except Exception as e:
                raise UserDefinedErrorHandlerError('Exception thrown in user-defined skipped request callback') from e

    def _get_message_from_error(self, error: Exception) -> str:
        """Get error message summary from exception.

        Custom processing to reduce the irrelevant traceback clutter in some cases.
        """
        traceback_parts = traceback.format_exception(type(error), value=error, tb=error.__traceback__, chain=True)
        used_traceback_parts = traceback_parts

        if (
            isinstance(error, asyncio.exceptions.TimeoutError)
            and traceback_parts
            and self._request_handler_timeout_text in traceback_parts[-1]
        ) or isinstance(error, UserHandlerTimeoutError):
            used_traceback_parts = reduce_asyncio_timeout_error_to_relevant_traceback_parts(error)
            used_traceback_parts.extend(traceback_parts[-1:])

        return ''.join(used_traceback_parts).strip('\n')

    def _get_only_inner_most_exception(self, error: BaseException) -> BaseException:
        """Get innermost exception by following __cause__ and __context__ attributes of exception."""
        if error.__cause__:
            return self._get_only_inner_most_exception(error.__cause__)
        if error.__context__:
            return self._get_only_inner_most_exception(error.__context__)
        # No __cause__ and no __context__, this is as deep as it can get.
        return error

    def _prepare_send_request_function(
        self,
        session: Session | None,
        proxy_info: ProxyInfo | None,
    ) -> SendRequestFunction:
        async def send_request(
            url: str,
            *,
            method: HttpMethod = 'GET',
            payload: HttpPayload | None = None,
            headers: HttpHeaders | dict[str, str] | None = None,
        ) -> HttpResponse:
            return await self._http_client.send_request(
                url=url,
                method=method,
                payload=payload,
                headers=headers,
                session=session,
                proxy_info=proxy_info,
            )

        return send_request

    def _convert_url_to_request_iterator(self, urls: Sequence[str | Request], base_url: str) -> Iterator[Request]:
        """Convert a sequence of URLs or Request objects to an iterator of Request objects."""
        for url in urls:
            # If the request is a Request object, keep it as it is
            if isinstance(url, Request):
                yield url
            # If the request is a string, convert it to Request object with absolute_url.
            elif isinstance(url, str) and not is_url_absolute(url):
                absolute_url = convert_to_absolute_url(base_url, url)
                yield Request.from_url(absolute_url)
            else:
                yield Request.from_url(url)

    async def _add_requests(
        self,
        context: BasicCrawlingContext,
        requests: Sequence[str | Request],
        rq_id: str | None = None,
        rq_name: str | None = None,
        rq_alias: str | None = None,
        **kwargs: Unpack[EnqueueLinksKwargs],
    ) -> None:
        """Add requests method aware of the crawling context."""
        if rq_id or rq_name or rq_alias:
            request_manager: RequestManager = await RequestQueue.open(
                id=rq_id,
                name=rq_name,
                alias=rq_alias,
                storage_client=self._service_locator.get_storage_client(),
                configuration=self._service_locator.get_configuration(),
            )
        else:
            request_manager = await self.get_request_manager()

        context_aware_requests = list[Request]()
        base_url = kwargs.get('base_url') or context.request.loaded_url or context.request.url
        requests_iterator = self._convert_url_to_request_iterator(requests, base_url)
        filter_requests_iterator = self._enqueue_links_filter_iterator(requests_iterator, context.request.url, **kwargs)
        for dst_request in filter_requests_iterator:
            # Update the crawl depth of the request.
            dst_request.crawl_depth = context.request.crawl_depth + 1

            if self._max_crawl_depth is None or dst_request.crawl_depth <= self._max_crawl_depth:
                context_aware_requests.append(dst_request)

        return await request_manager.add_requests(context_aware_requests)

    async def _commit_request_handler_result(self, context: BasicCrawlingContext) -> None:
        """Commit request handler result for the input `context`. Result is taken from `_context_result_map`."""
        result = self._context_result_map[context]

        for add_requests_call in result.add_requests_calls:
            await self._add_requests(context, **add_requests_call)

        for push_data_call in result.push_data_calls:
            await self._push_data(**push_data_call)

        await self._commit_key_value_store_changes(result, get_kvs=self.get_key_value_store)

        result.apply_request_changes(target=context.request)

    @staticmethod
    async def _commit_key_value_store_changes(
        result: RequestHandlerRunResult, get_kvs: GetKeyValueStoreFromRequestHandlerFunction
    ) -> None:
        """Store key value store changes recorded in result."""
        for (id, name, alias), changes in result.key_value_store_changes.items():
            store = await get_kvs(id=id, name=name, alias=alias)
            for key, value in changes.updates.items():
                await store.set_value(key, value.content, value.content_type)

    async def __is_finished_function(self) -> bool:
        self._stop_if_max_requests_count_exceeded()
        if self._unexpected_stop:
            self._logger.info('The crawler will finish any remaining ongoing requests and shut down.')
            return True

        if self._abort_on_error and self._failed:
            self._failed = False
            return True

        if self._keep_alive:
            return False

        request_manager = await self.get_request_manager()
        return await request_manager.is_finished()

    async def __is_task_ready_function(self) -> bool:
        self._stop_if_max_requests_count_exceeded()
        if self._unexpected_stop:
            self._logger.info(
                'No new requests are allowed because crawler `stop` method was called. '
                'Ongoing requests will be allowed to complete.'
            )
            return False

        request_manager = await self.get_request_manager()
        return not await request_manager.is_empty()

    async def __run_task_function(self) -> None:
        request_manager = await self.get_request_manager()

        request = await wait_for(
            request_manager.fetch_next_request,
            timeout=self._internal_timeout,
            timeout_message=f'Fetching next request failed after {self._internal_timeout.total_seconds()} seconds',
            logger=self._logger,
            max_retries=3,
        )

        if request is None:
            return

        if not (await self._is_allowed_based_on_robots_txt_file(request.url)):
            self._logger.warning(
                f'Skipping request {request.url} ({request.unique_key}) because it is disallowed based on robots.txt'
            )

            await self._handle_skipped_request(request, 'robots_txt', need_mark=True)
            return

        if request.session_id:
            session = await self._get_session_by_id(request.session_id)
        else:
            session = await self._get_session()
        proxy_info = await self._get_proxy_info(request, session)
        result = RequestHandlerRunResult(key_value_store_getter=self.get_key_value_store, request=request)

        context = BasicCrawlingContext(
            request=result.request,
            session=session,
            proxy_info=proxy_info,
            send_request=self._prepare_send_request_function(session, proxy_info),
            add_requests=result.add_requests,
            push_data=result.push_data,
            get_key_value_store=result.get_key_value_store,
            use_state=self._use_state,
            log=self._logger,
        )
        self._context_result_map[context] = result

        self._statistics.record_request_processing_start(request.unique_key)

        try:
            request.state = RequestState.REQUEST_HANDLER

            try:
                with swapped_context(context, request):
                    self._check_request_collision(request, session)
                    await self._run_request_handler(context=context)
            except asyncio.TimeoutError as e:
                raise RequestHandlerError(e, context) from e

            await self._commit_request_handler_result(context)

            request.state = RequestState.DONE

            await self._mark_request_as_handled(request)

            # Record successful request to reset rate limit backoff for this domain.
            if isinstance(request_manager, ThrottlingRequestManager):
                request_manager.record_success(request.url)

            if session and session.is_usable:
                session.mark_good()

            self._statistics.record_request_processing_finish(request.unique_key)

        except RequestCollisionError as request_error:
            request.no_retry = True
            await self._handle_request_error(context, request_error)

        except RequestHandlerError as primary_error:
            primary_error = cast(
                'RequestHandlerError[TCrawlingContext]', primary_error
            )  # valid thanks to ContextPipeline

            self._logger.debug(
                'An exception occurred in the user-defined request handler',
                exc_info=primary_error.wrapped_exception,
            )
            await self._handle_request_error(primary_error.crawling_context, primary_error.wrapped_exception)

        except SessionError as session_error:
            if not session:
                raise RuntimeError('SessionError raised in a crawling context without a session') from session_error

            if self._error_handler:
                await self._error_handler(context, session_error)

            if self._should_retry_request(context, session_error):
                exc_only = ''.join(traceback.format_exception_only(session_error)).strip()
                self._logger.warning('Encountered "%s", rotating session and retrying...', exc_only)

                if session:
                    session.retire()

                # Increment session rotation count.
                request.session_rotation_count = (request.session_rotation_count or 0) + 1

                await request_manager.reclaim_request(request)
                await self._statistics.error_tracker_retry.add(error=session_error, context=context)
            else:
                await self._mark_request_as_handled(request)

                await self._handle_failed_request(context, session_error)
                self._statistics.record_request_processing_failure(request.unique_key)

        except ContextPipelineInterruptedError as interrupted_error:
            self._logger.debug('The context pipeline was interrupted', exc_info=interrupted_error)

            await self._mark_request_as_handled(request)

        except ContextPipelineInitializationError as initialization_error:
            self._logger.debug(
                'An exception occurred during the initialization of crawling context',
                exc_info=initialization_error,
            )
            await self._handle_request_error(context, initialization_error.wrapped_exception)

        except Exception as internal_error:
            self._logger.exception(
                'An exception occurred during handling of a request. This places the crawler '
                'and its underlying storages into an unknown state and crawling will be terminated.',
                exc_info=internal_error,
            )
            raise

    async def _run_request_handler(self, context: BasicCrawlingContext) -> None:
        context.request.state = RequestState.BEFORE_NAV
        await self._context_pipeline(
            context,
            lambda final_context: wait_for(
                lambda: self.router(final_context),
                timeout=self._request_handler_timeout,
                timeout_message=f'{self._request_handler_timeout_text}'
                f' {self._request_handler_timeout.total_seconds()} seconds',
                logger=self._logger,
            ),
        )

    def _raise_for_error_status_code(self, status_code: int) -> None:
        """Raise an exception if the given status code is considered an error.

        Args:
            status_code: The HTTP status code to check.

        Raises:
            HttpStatusCodeError: If the status code represents a server error or is explicitly configured as an error.
            HttpClientStatusCodeError: If the status code represents a client error.
        """
        is_ignored_status = status_code in self._ignore_http_error_status_codes
        is_explicit_error = status_code in self._additional_http_error_status_codes

        if is_explicit_error:
            raise HttpStatusCodeError('Error status code (user-configured) returned.', status_code)

        if is_status_code_client_error(status_code) and not is_ignored_status:
            raise HttpClientStatusCodeError('Client error status code returned', status_code)

        if is_status_code_server_error(status_code) and not is_ignored_status:
            raise HttpStatusCodeError('Error status code returned', status_code)

    def _raise_for_session_blocked_status_code(
        self,
        session: Session | None,
        status_code: int,
        *,
        request_url: str = '',
        retry_after_header: str | None = None,
    ) -> None:
        """Raise an exception if the given status code indicates the session is blocked.

        If the status code is 429 (Too Many Requests), the domain is recorded as
        rate-limited in the `ThrottlingRequestManager` for per-domain backoff.

        Args:
            session: The session used for the request. If None, no check is performed.
            status_code: The HTTP status code to check.
            request_url: The request URL, used for per-domain rate limit tracking.
            retry_after_header: The value of the Retry-After response header, if present.

        Raises:
            SessionError: If the status code indicates the session is blocked.
        """
        if status_code == 429 and request_url:  # noqa: PLR2004
            retry_after = parse_retry_after_header(retry_after_header)

            # _request_manager might not be initialized yet if called directly or early,
            # but usually it's set in get_request_manager().
            if isinstance(self._request_manager, ThrottlingRequestManager):
                self._request_manager.record_domain_delay(request_url, retry_after=retry_after)

        if session is not None and session.is_blocked_status_code(
            status_code=status_code,
            ignore_http_error_status_codes=self._ignore_http_error_status_codes,
        ):
            raise SessionError(f'Assuming the session is blocked based on HTTP status code {status_code}')

    # NOTE: _parse_retry_after_header has been moved to crawlee._utils.http.parse_retry_after_header
    def _check_request_collision(self, request: Request, session: Session | None) -> None:
        """Raise an exception if a request cannot access required resources.

        Args:
            request: The `Request` that might require specific resources (like a session).
            session: The `Session` that was retrieved for the request, or `None` if not available.

        Raises:
            RequestCollisionError: If the `Session` referenced by the `Request` is not available.
        """
        if self._use_session_pool and request.session_id and not session:
            raise RequestCollisionError(
                f'The Session (id: {request.session_id}) bound to the Request is no longer available in SessionPool'
            )

    async def _is_allowed_based_on_robots_txt_file(self, url: str) -> bool:
        """Check if the URL is allowed based on the robots.txt file.

        Args:
            url: The URL to verify against robots.txt rules. Returns True if crawling this URL is permitted.
        """
        if not self._respect_robots_txt_file:
            return True
        robots_txt_file = await self._get_robots_txt_file_for_url(url)
        if not robots_txt_file:
            return True

        # Wire robots.txt crawl-delay into ThrottlingRequestManager
        if isinstance(self._request_manager, ThrottlingRequestManager):
            crawl_delay = robots_txt_file.get_crawl_delay()
            if crawl_delay is not None:
                self._request_manager.set_crawl_delay(url, crawl_delay)

        return robots_txt_file.is_allowed(url)

    async def _get_robots_txt_file_for_url(self, url: str) -> RobotsTxtFile | None:
        """Get the RobotsTxtFile for a given URL.

        Args:
            url: The URL whose domain will be used to locate and fetch the corresponding robots.txt file.
        """
        if not self._respect_robots_txt_file:
            return None
        origin_url = str(URL(url).origin())
        robots_txt_file = self._robots_txt_file_cache.get(origin_url)
        if robots_txt_file:
            return robots_txt_file

        async with self._robots_txt_lock:
            # Check again if the robots.txt file is already cached after acquiring the lock
            robots_txt_file = self._robots_txt_file_cache.get(origin_url)
            if robots_txt_file:
                return robots_txt_file

            # If not cached, fetch the robots.txt file
            robots_txt_file = await self._find_txt_file_for_url(url)
            self._robots_txt_file_cache[origin_url] = robots_txt_file
            return robots_txt_file

    async def _find_txt_file_for_url(self, url: str) -> RobotsTxtFile:
        """Find the robots.txt file for a given URL.

        Args:
            url: The URL whose domain will be used to locate and fetch the corresponding robots.txt file.
        """
        return await RobotsTxtFile.find(url, self._http_client)

    def _log_status_message(self, message: str, level: LogLevel = 'DEBUG') -> None:
        """Log a status message for the crawler.

        Args:
            message: The status message to log.
            level: The logging level for the message.
        """
        log_level = string_to_log_level(level)
        self.log.log(log_level, message)

    async def _crawler_state_task(self) -> None:
        """Emit a persist state event with the given migration status."""
        event_manager = self._service_locator.get_event_manager()

        current_state = self.statistics.state

        if (
            failed_requests := (
                current_state.requests_failed - (self._previous_crawler_state or current_state).requests_failed
            )
            > 0
        ):
            message = f'Experiencing problems, {failed_requests} failed requests since last status update.'
        else:
            request_manager = await self.get_request_manager()
            total_count = await request_manager.get_total_count()
            if total_count is not None and total_count > 0:
                pages_info = f'{self._statistics.state.requests_finished}/{total_count}'
            else:
                pages_info = str(self._statistics.state.requests_finished)

            message = (
                f'Crawled {pages_info} pages, {self._statistics.state.requests_failed} failed requests, '
                f'desired concurrency {self._autoscaled_pool.desired_concurrency}.'
            )

        if self._status_message_callback:
            new_message = await self._status_message_callback(current_state, self._previous_crawler_state, message)
            if new_message:
                message = new_message
                self._log_status_message(message, level='INFO')
        else:
            self._log_status_message(message, level='INFO')

        event_manager.emit(
            event=Event.CRAWLER_STATUS, event_data=EventCrawlerStatusData(message=message, crawler_id=id(self))
        )

        self._previous_crawler_state = current_state

    async def _mark_request_as_handled(self, request: Request) -> None:
        request_manager = await self.get_request_manager()
        await wait_for(
            lambda: request_manager.mark_request_as_handled(request),
            timeout=self._internal_timeout,
            timeout_message='Marking request as handled timed out after '
            f'{self._internal_timeout.total_seconds()} seconds',
            logger=self._logger,
            max_retries=3,
        )