Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 32 additions & 0 deletions docs/spiders/advanced.md
Original file line number Diff line number Diff line change
Expand Up @@ -216,6 +216,37 @@ There's no automatic expiration. To force a fresh crawl, delete the cache direct

Development mode is meant for development, not production. Cached responses never expire, and replay bypasses rate limiting and blocked-request retries. Don't ship a spider with `development_mode = True`.

## Conditional Requests

Spiders that run on a schedule, like one that checks a set of feeds every few minutes, mostly download pages that haven't changed since the last run. With `conditional_requests = True`, the spider saves the `ETag` and `Last-Modified` headers of each page it processed, and the next run sends them back as `If-None-Match` / `If-Modified-Since`. A website that supports them answers `304 Not Modified` with an empty body instead of the whole page.

```python
class FeedMonitor(Spider):
name = "feed_monitor"
start_urls = ["https://example.com/feed.xml"]
conditional_requests = True

async def parse(self, response: Response):
for item in response.css("item"):
yield {"title": item.css("title::text").get("")}

async def on_not_modified(self, request, response):
self.logger.info(f"No changes in {request.url}")
```

When the answer is `304`, the request's callback is skipped, since there's nothing new for it to process, and the `on_not_modified()` hook is called instead. The `not_modified_count` stat counts these responses, and AutoThrottle treats them as healthy ones.

### How It Works

1. **Which requests**: GET and HEAD requests made with HTTP sessions (`FetcherSession`). Browser sessions fetch normally, and requests that already set their own `If-None-Match` / `If-Modified-Since` headers are left as they are.
2. **When validators are saved**: Only after the callback of a `2xx` response finished without raising. A page whose callback failed, or that was blocked, is downloaded in full on the next run.
3. **Following links**: The conditional headers are sent for that single fetch only, so requests created with `response.follow()` don't inherit them.
4. **Storage**: One small JSON file per request fingerprint in `.scrapling_validators/{spider.name}/`, or in the directory set with `conditional_requests_dir`. Delete it to make the next run download everything again.

!!! note

Conditional requests are disabled while `development_mode` is on, since the response cache already replays pages without fetching them, and a `304` would leave it nothing to cache.

## Streaming

For long-running spiders or applications that need real-time access to scraped items, use the `stream()` method instead of `start()`:
Expand Down Expand Up @@ -357,6 +388,7 @@ print(f"Offsite filtered: {stats.offsite_requests_count}")
print(f"Robots.txt disallowed: {stats.robots_disallowed_count}")
print(f"Cache hits: {stats.cache_hits}")
print(f"Cache misses: {stats.cache_misses}")
print(f"Not modified: {stats.not_modified_count}")
print(f"Items scraped: {stats.items_scraped}")
print(f"Items dropped: {stats.items_dropped}")
print(f"Response bytes: {stats.response_bytes}")
Expand Down
83 changes: 83 additions & 0 deletions scrapling/spiders/conditional.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,83 @@
from pathlib import Path

import orjson
import anyio
from anyio import Path as AsyncPath

from scrapling.core.utils import log
from scrapling.core._types import Any, Dict, Mapping, Optional
from scrapling.engines.toolbelt.custom import Response

# Headers that make a request conditional; a request that already sends one is left to its author
CONDITIONAL_HEADERS = frozenset({"if-none-match", "if-modified-since", "if-match", "if-unmodified-since", "if-range"})


def _get_header(headers: Mapping[str, Any], name: str) -> Optional[str]:
"""Case-insensitive header lookup, since engines don't agree on the case of header names."""
for key, value in headers.items():
if key.lower() == name:
return str(value)
return None


class ConditionalRequestManager:
"""Saves the validators (`ETag` / `Last-Modified`) of processed responses to disk, so the next run can ask the
website whether each page changed instead of downloading it again."""

def __init__(self, validators_dir: str | Path):
self._validators_dir = AsyncPath(validators_dir)

def _validators_path(self, fingerprint: bytes) -> AsyncPath:
return self._validators_dir / f"{fingerprint.hex()}.json"

async def get_headers(self, fingerprint: bytes) -> Dict[str, str]:
"""Return the `If-None-Match` / `If-Modified-Since` headers for a request, empty if nothing was saved for it."""
path = self._validators_path(fingerprint)
if not await path.exists():
return {}

try:
async with await anyio.open_file(path, "rb") as f:
data: Dict[str, Any] = orjson.loads(await f.read())
except Exception as e:
log.warning(f"Failed to read saved validators for {fingerprint.hex()}: {e}")
return {}

headers = {}
if data.get("etag"):
headers["If-None-Match"] = data["etag"]
if data.get("last_modified"):
headers["If-Modified-Since"] = data["last_modified"]
return headers

async def put(self, fingerprint: bytes, response: Response) -> None:
"""Save the response's validators, or drop the saved ones when the website stopped sending any."""
path = self._validators_path(fingerprint)
headers = response.headers or {}
etag = _get_header(headers, "etag")
last_modified = _get_header(headers, "last-modified")
if not etag and not last_modified:
if await path.exists():
await path.unlink()
return

await self._validators_dir.mkdir(parents=True, exist_ok=True)
temp_path = path.with_suffix(".tmp")
try:
serialized = orjson.dumps({"url": response.url, "etag": etag, "last_modified": last_modified})
async with await anyio.open_file(temp_path, "wb") as f:
await f.write(serialized)

await temp_path.replace(path)
except Exception as e:
if await temp_path.exists():
await temp_path.unlink()
log.warning(f"Failed to save validators for {fingerprint.hex()}: {e}")

async def clear(self) -> None:
if not await self._validators_dir.exists():
return
async for entry in self._validators_dir.iterdir():
if entry.suffix == ".json":
await entry.unlink()
log.info(f"Cleared saved validators at {self._validators_dir}")
53 changes: 49 additions & 4 deletions scrapling/spiders/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
from scrapling.spiders.result import CrawlStats, ItemList
from scrapling.spiders.throttle import AutoThrottle, parse_retry_after
from scrapling.spiders.cache import ResponseCacheManager
from scrapling.spiders.conditional import ConditionalRequestManager, CONDITIONAL_HEADERS
from scrapling.spiders.checkpoint import CheckpointManager, CheckpointData
from scrapling.core._types import Dict, Union, Optional, TYPE_CHECKING, Any, AsyncGenerator

Expand Down Expand Up @@ -62,6 +63,15 @@ async def _fetch_robots(url: str, sid: str) -> Response:
else:
self._cache_manager = None

self._conditional_manager: Optional[ConditionalRequestManager] = None
if getattr(self.spider, "conditional_requests", False):
if self._cache_manager:
# A cache miss needs the full body to cache it, so a 304 would leave nothing to replay
log.warning("Conditional requests are disabled in development mode, the response cache is used instead")
else:
validators_dir = self.spider.conditional_requests_dir or f".scrapling_validators/{self.spider.name}"
self._conditional_manager = ConditionalRequestManager(validators_dir)

if self.spider.autothrottle_enabled:
self._autothrottle: Optional[AutoThrottle] = AutoThrottle(
start_delay=self.spider.autothrottle_start_delay,
Expand Down Expand Up @@ -152,8 +162,19 @@ def _normalize_request(self, request: Request) -> None:
if not request.sid:
request.sid = self.session_manager.default_session_id

async def _run_callbacks(self, request: Request, response: Response) -> None:
"""Dispatch response to the request's callback and process yielded items/requests."""
@staticmethod
def _uses_validators(request: Request) -> bool:
"""Whether conditional requests apply to this request: only GET/HEAD ones that don't set their own validators."""
if str(request._session_kwargs.get("method", "GET")).upper() not in ("GET", "HEAD"):
return False
headers = request._session_kwargs.get("headers") or {}
return not any(key.lower() in CONDITIONAL_HEADERS for key in headers)

async def _run_callbacks(self, request: Request, response: Response) -> bool:
"""Dispatch response to the request's callback and process yielded items/requests.

Returns whether the callback finished without raising.
"""
callback = request.callback if request.callback else self.spider.parse
try:
async for result in callback(response):
Expand Down Expand Up @@ -182,6 +203,8 @@ async def _run_callbacks(self, request: Request, response: Response) -> None:
msg = f"Spider error processing {request}:\n {e}"
log.error(msg, exc_info=e)
await self.spider.on_error(request, e)
return False
return True

async def _process_request(self, request: Request) -> None:
"""Download and process a single request."""
Expand Down Expand Up @@ -211,6 +234,10 @@ async def _process_request(self, request: Request) -> None:
await self._run_callbacks(request, cached)
return

conditional_headers: Dict[str, str] = {}
if self._conditional_manager and request._fp is not None and self._uses_validators(request):
conditional_headers = await self._conditional_manager.get_headers(request._fp)

async with self._rate_limiter(request.domain):
if self._autothrottle:
delay = self._autothrottle.delay_for(request.domain, floor)
Expand All @@ -224,7 +251,7 @@ async def _process_request(self, request: Request) -> None:
self.stats.proxies.append(dict(request._session_kwargs["proxies"]))
try:
started_at = anyio.current_time()
response = await self.session_manager.fetch(request)
response = await self.session_manager.fetch(request, headers=conditional_headers)
latency = anyio.current_time() - started_at
self.stats.increment_requests_count(request.sid or self.session_manager.default_session_id)
self.stats.increment_response_bytes(request.domain, len(response.body))
Expand All @@ -238,6 +265,15 @@ async def _process_request(self, request: Request) -> None:
if self._cache_manager and request._fp is not None:
self.stats.cache_misses += 1

if conditional_headers and response.status == 304:
# Nothing changed since a previous run processed this page, so the callback has nothing new to work on
self.stats.not_modified_count += 1
if self._autothrottle:
self._autothrottle.record(request.domain, latency, True, floor)
log.debug(f"Not modified since the last run: {request.url}")
await self.spider.on_not_modified(request, response)
return

blocked = await self.spider.is_blocked(response)
if self._autothrottle:
ok = 200 <= response.status < 300 and not blocked
Expand Down Expand Up @@ -267,7 +303,16 @@ async def _process_request(self, request: Request) -> None:
if self._cache_manager and request._fp is not None:
await self._cache_manager.put(request._fp, response, request._session_kwargs.get("method", "GET"))

await self._run_callbacks(request, response)
processed = await self._run_callbacks(request, response)
# Validators are saved only once the callback succeeded, so a page that failed is downloaded in full next time
if (
processed
and self._conditional_manager
and request._fp is not None
and 200 <= response.status < 300
and self._uses_validators(request)
):
await self._conditional_manager.put(request._fp, response)

async def _task_wrapper(self, request: Request) -> None:
"""Wrapper to track active task count."""
Expand Down
2 changes: 2 additions & 0 deletions scrapling/spiders/result.py
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,7 @@ class CrawlStats:
robots_disallowed_count: int = 0
cache_hits: int = 0
cache_misses: int = 0
not_modified_count: int = 0
response_bytes: int = 0
items_scraped: int = 0
items_dropped: int = 0
Expand Down Expand Up @@ -185,6 +186,7 @@ def to_dict(self) -> dict[str, Any]:
"robots_disallowed_count": self.robots_disallowed_count,
"cache_hits": self.cache_hits,
"cache_misses": self.cache_misses,
"not_modified_count": self.not_modified_count,
"blocked_requests_count": self.blocked_requests_count,
"response_status_count": self.response_status_count,
"response_bytes": self.response_bytes,
Expand Down
18 changes: 14 additions & 4 deletions scrapling/spiders/session.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from scrapling.spiders.request import Request
from scrapling.engines.static import _ASyncSessionLogic
from scrapling.engines.toolbelt.convertor import Response
from scrapling.core._types import Set, cast, SUPPORTED_HTTP_METHODS
from scrapling.core._types import Dict, Optional, Set, cast, SUPPORTED_HTTP_METHODS
from scrapling.fetchers import AsyncStealthySession, FetcherSession

Session = FetcherSession | AsyncStealthySession
Expand Down Expand Up @@ -100,7 +100,13 @@ async def close(self) -> None:

self._started = False

async def fetch(self, request: Request) -> Response:
async def fetch(self, request: Request, headers: Optional[Dict[str, str]] = None) -> Response:
"""Fetch a request with the session its `sid` points to.

:param request: The request to fetch.
:param headers: Headers to add for this fetch only. They aren't saved on the request, so requests followed
from the response don't inherit them. Browser sessions read `extra_headers` instead, so they ignore these.
"""
sid = request.sid if request.sid else self.default_session_id
session = self.get(sid)

Expand All @@ -110,11 +116,15 @@ async def fetch(self, request: Request) -> Response:
if not session._is_alive:
await session.__aenter__()

session_kwargs = request._session_kwargs
if headers:
session_kwargs = {**session_kwargs, "headers": {**(session_kwargs.get("headers") or {}), **headers}}

if isinstance(session, FetcherSession):
client = session._client

if isinstance(client, _ASyncSessionLogic):
kwargs = request._session_kwargs.copy()
kwargs = session_kwargs.copy()
method = cast(SUPPORTED_HTTP_METHODS, kwargs.pop("method", "GET"))
if "stealthy_headers" in kwargs:
# Same mapping as the public session methods (`get`, `post`, ...), so it doesn't reach curl_cffi
Expand All @@ -128,7 +138,7 @@ async def fetch(self, request: Request) -> Response:
# Sync session or other types - shouldn't happen in async context
raise TypeError(f"Session type {type(client)} not supported for async fetch")
else:
response = await session.fetch(url=request.url, **request._session_kwargs)
response = await session.fetch(url=request.url, **session_kwargs)

response.request = request
# Merge request meta into response meta (response meta takes priority)
Expand Down
12 changes: 12 additions & 0 deletions scrapling/spiders/spider.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,10 @@ class Spider(ABC):
development_mode: bool = False
development_cache_dir: Optional[str] = None

# Conditional requests (ETag / Last-Modified validators saved between runs)
conditional_requests: bool = False
conditional_requests_dir: Optional[str] = None

# Concurrency settings
concurrent_requests: int = 4
concurrent_requests_per_domain: int = 0
Expand Down Expand Up @@ -197,6 +201,14 @@ async def on_error(self, request: Request, error: Exception) -> None:
"""
pass

async def on_not_modified(self, request: Request, response: "Response") -> None:
"""Called instead of the request's callback when `conditional_requests` is enabled and the website answered
`304 Not Modified`, meaning the page didn't change since a previous run processed it.

Override to act on unchanged pages.
"""
pass

async def on_scraped_item(self, item: Dict[str, Any]) -> Dict[str, Any] | None:
"""A hook to be overridden by users to do some processing on scraped items, return `None` to drop the item silently."""
return item
Expand Down
Loading