Spider Architecture

from scrapling.spiders import Spider, Response, Request

class MySpider(Spider):
    name = "my_spider"
    start_urls = ["https://example.com"]

    async def parse(self, response: Response):
        for link in response.css("a::attr(href)").getall():
            yield response.follow(link, callback=self.parse_page)

    async def parse_page(self, response: Response):
        yield {"title": response.css("h1::text").get("")}

async def enqueue(self, request: Request) -> bool:
    """Add a request to the queue."""
    fingerprint = request.update_fingerprint(self._include_kwargs, self._include_headers, self._keep_fragments)

    if not request.dont_filter and fingerprint in self._seen:
        log.debug("Dropped duplicate request: %s", request)
        return False

    self._seen.add(fingerprint)

    # Negative priority so higher priority = dequeued first
    counter = next(self._counter)
    item = (-request.priority, counter, request)
    self._pending[counter] = item
    await self._queue.put(item)
    return True

def add(self, session_id: str, session: Session, *, default: bool = False, lazy: bool = False) -> "SessionManager":
    """Register a session instance.

    :param session_id: Name to reference this session in requests
    :param session: Your pre-configured session instance
    :param default: If True, this becomes the default session
    :param lazy: If True, the session will be started only when a request uses its ID.
    """
    if session_id in self._sessions:
        raise ValueError(f"Session '{session_id}' already registered")

    self._sessions[session_id] = session

    if default or self._default_session_id is None:
        self._default_session_id = session_id

    if lazy:
        self._lazy_sessions.add(session_id)

    return self

async def save(self, data: CheckpointData) -> None:
    """Save checkpoint data to disk atomically."""
    await self.crawldir.mkdir(parents=True, exist_ok=True)

    temp_path = self._checkpoint_path.with_suffix(".tmp")

    try:
        serialized = pickle.dumps(data, protocol=pickle.HIGHEST_PROTOCOL)
        async with await anyio.open_file(temp_path, "wb") as f:
            await f.write(serialized)

        await temp_path.rename(self._checkpoint_path)

        log.info(f"Checkpoint saved: {len(data.requests)} requests, {len(data.seen)} seen URLs")
    except Exception as e:
        # Clean up temp file if it exists
        if await temp_path.exists():
            await temp_path.unlink()
        log.error(f"Failed to save checkpoint: {e}")
        raise

class ItemList(list):
    """A list of scraped items with export capabilities."""

    def to_json(self, path: Union[str, Path], *, indent: bool = False):
        """Export items to a JSON file.

        :param path: Path to the output file
        :param indent: Pretty-print with 2-space indentation (slightly slower)
        """
        options = orjson.OPT_SERIALIZE_NUMPY
        if indent:
            options |= orjson.OPT_INDENT_2

        file = Path(path)
        file.parent.mkdir(parents=True, exist_ok=True)
        file.write_bytes(orjson.dumps(list(self), option=options))
        log.info("Saved %d items to %s", len(self), path)

    def to_jsonl(self, path: Union[str, Path]):
        """Export items as JSON Lines (one JSON object per line).

        :param path: Path to the output file
        """
        Path(path).parent.mkdir(parents=True, exist_ok=True)
        with open(path, "wb") as f:
            for item in self:
                f.write(orjson.dumps(item, option=orjson.OPT_SERIALIZE_NUMPY))
                f.write(b"\n")
        log.info("Saved %d items to %s", len(self), path)