feat(crawlers): implement Request.skip_navigation to skip navigation and run the handler directly

src/crawlee/_request.py

@@ -403,6 +403,20 @@ def enqueue_strategy(self) -> EnqueueStrategy:
     def enqueue_strategy(self, new_enqueue_strategy: EnqueueStrategy) -> None:
         self.crawlee_data.enqueue_strategy = new_enqueue_strategy
 
+    @property
+    def skip_navigation(self) -> bool:
+        """Tells the crawler processing this request to skip the navigation and process the request directly.
+
+        When this is set to `True`, the crawling context will not contain the results of the navigation
+        (e.g. the HTTP response, parsed content or `loaded_url`). Accessing these properties will raise
+        a `NavigationSkippedError` at runtime.
+        """
+        return self.crawlee_data.skip_navigation
+
+    @skip_navigation.setter
+    def skip_navigation(self, new_value: bool) -> None:
+        self.crawlee_data.skip_navigation = new_value
+
     @property
     def was_already_handled(self) -> bool:
         """Indicates whether the request was handled."""

src/crawlee/crawlers/_basic/_basic_crawler.py

@@ -12,6 +12,7 @@
 from asyncio import CancelledError
 from collections.abc import AsyncGenerator, Awaitable, Callable, Iterable, Sequence
 from contextlib import AsyncExitStack
+from dataclasses import replace as dataclass_replace
 from datetime import timedelta
 from functools import partial
 from http import HTTPStatus
@@ -57,6 +58,7 @@
     ContextPipelineInterruptedError,
     HttpClientStatusCodeError,
     HttpStatusCodeError,
+    NavigationSkippedError,
     PersistentRateLimitError,
     RequestCollisionError,
     RequestHandlerError,
@@ -115,6 +117,33 @@
 SkippedRequestCallback = Callable[[str, SkippedReason], Awaitable[None]]
 
 
+class _NavigationSkippedRequestProxy:
+    """Wraps a `Request` whose navigation was skipped due to `skip_navigation`.
+
+    All attribute access is forwarded to the wrapped request, except `loaded_url`, which raises a
+    `NavigationSkippedError` — mirroring the JS SDK behavior for skipped navigations.
+    """
+
+    __slots__ = ('_raise_on_loaded_url', '_request')
+
+    def __init__(self, request: Request, raise_on_loaded_url: Callable[[], str]) -> None:
+        self._request = request
+        self._raise_on_loaded_url = raise_on_loaded_url
+
+    def __getattr__(self, name: str) -> Any:
+        # Attribute writes (e.g. `request.state = ...` in the router) must land on the wrapped request,
+        # so intercept only reads that the wrapped request cannot resolve itself.
+        if name == 'loaded_url':
+            return self._raise_on_loaded_url()
+        return getattr(self._request, name)
+
+    def __setattr__(self, name: str, value: Any) -> None:
+        if name in ('_request', '_raise_on_loaded_url'):
+            super().__setattr__(name, value)
+        else:
+            setattr(self._request, name, value)
+
+
 class _BasicCrawlerOptions(TypedDict):
     """Non-generic options the `BasicCrawler` constructor."""
 
@@ -1722,6 +1751,11 @@ async def _defer_throttled_request(self, request: Request, error: RequestThrottl
 
     async def _run_request_handler(self, context: BasicCrawlingContext) -> None:
         context.request.state = RequestState.BEFORE_NAV
+
+        if context.request.skip_navigation:
+            await self._run_request_handler_with_skipped_navigation(context)
+            return
+
         await self._context_pipeline(
             context,
             lambda final_context: wait_for(
@@ -1733,6 +1767,37 @@ async def _run_request_handler(self, context: BasicCrawlingContext) -> None:
             ),
         )
 
+    async def _run_request_handler_with_skipped_navigation(self, context: BasicCrawlingContext) -> None:
+        """Run the request handler without navigating, for requests with `skip_navigation` set.
+
+        The context pipeline is bypassed entirely: its middleware steps are responsible for the navigation,
+        so the crawling context they would produce does not contain the navigation results. Navigation-only
+        context members (`request.loaded_url`) raise a `NavigationSkippedError` instead of returning `None`.
+        """
+        self._logger.debug(f'Skipping navigation for request {context.request.url} (`skip_navigation` was used).')
+
+        def _raise_on_loaded_url() -> str:
+            raise NavigationSkippedError(
+                'The `request.loaded_url` property is not available - `skip_navigation` was used.'
+            )
+
+        guarded_request = cast(
+            'Request',
+            _NavigationSkippedRequestProxy(context.request, _raise_on_loaded_url),
+        )
+        guarded_context = cast(
+            'TCrawlingContext',
+            dataclass_replace(context, request=guarded_request),  # ty: ignore[invalid-argument-type]
+        )
+
+        await wait_for(
+            lambda: self.router(guarded_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.
 

src/crawlee/errors.py

@@ -13,6 +13,7 @@
     'ContextPipelineInterruptedError',
     'HttpClientStatusCodeError',
     'HttpStatusCodeError',
+    'NavigationSkippedError',
     'PersistentRateLimitError',
     'ProxyError',
     'RequestCollisionError',
@@ -35,6 +36,16 @@ class UserHandlerTimeoutError(UserDefinedErrorHandlerError):
     """Raised when a router fails due to user raised timeout. This is different from user-defined handler timing out."""
 
 
+@docs_group('Errors')
+class NavigationSkippedError(Exception):
+    """Raised when a navigation-only crawling context member is accessed, but `skip_navigation` was used.
+
+    When a request has `skip_navigation` set to `True`, the crawler does not open the page and the crawling
+    context does not contain the results of the navigation (e.g. `http_response`, `$`, parsed content or
+    `request.loaded_url`). Accessing such properties raises this error at runtime.
+    """
+
+
 @docs_group('Errors')
 class SessionError(Exception):
     """Errors of `SessionError` type will trigger a session rotation.

tests/unit/crawlers/_basic/test_basic_crawler.py

@@ -30,6 +30,7 @@
 from crawlee.configuration import Configuration
 from crawlee.crawlers import BasicCrawler
 from crawlee.errors import (
+    NavigationSkippedError,
     PersistentRateLimitError,
     RequestCollisionError,
     RequestThrottledError,
@@ -3169,3 +3170,45 @@ async def test_stall_ignores_shared_concurrency() -> None:
         pytest.raises(PersistentRateLimitError),
     ):
         await cast('Any', crawler)._BasicCrawler__is_finished_function()
+
+
+async def test_skip_navigation_runs_handler_without_navigation() -> None:
+    """A request with `skip_navigation` set is processed directly, without the navigation pipeline."""
+    crawler = BasicCrawler(
+        configure_logging=False,
+        concurrency_settings=ConcurrencySettings(desired_concurrency=1, max_concurrency=1),
+    )
+    processed: list[str] = []
+    navigation_members: dict[str, str] = {}
+
+    @crawler.router.default_handler
+    async def handler(context: BasicCrawlingContext) -> None:
+        processed.append(context.request.unique_key)
+        try:
+            _ = context.request.loaded_url
+        except NavigationSkippedError:
+            navigation_members[context.request.unique_key] = 'raised'
+
+    plain_request = Request.from_url('https://plain.placeholder.com', unique_key='plain')
+    skipped_request = Request.from_url('https://skipped.placeholder.com', unique_key='skipped')
+    skipped_request.skip_navigation = True
+
+    stats = await crawler.run([plain_request, skipped_request])
+
+    assert stats.requests_finished == 2
+    assert processed == ['plain', 'skipped']
+    # `loaded_url` only raises for the skip-navigation request; on the plain one it is just `None`.
+    assert navigation_members == {'skipped': 'raised'}
+
+
+async def test_skip_navigation_request_serialization_roundtrip() -> None:
+    """`skip_navigation` survives serialization via the `skipNavigation` user data alias."""
+    request = Request.from_url('https://placeholder.com', unique_key='roundtrip')
+    assert request.skip_navigation is False
+
+    request.skip_navigation = True
+    data = request.model_dump(by_alias=True)
+    assert data['userData']['__crawlee']['skipNavigation'] is True
+
+    restored = Request.model_validate(data)
+    assert restored.skip_navigation is True