Async Web Scraper with Rate Limiting, Retries, and Structured Data Extraction

Project Structure

scraper/
├── models.py          # Typed dataclasses
├── rate_limiter.py    # Token bucket rate limiter
├── retry.py           # Exponential backoff logic
├── scraper.py         # Core async scraper
├── extractors.py      # Data extraction & parsing
└── main.py            # Example usage

---

models.py — Typed Dataclasses

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, List, Any, Dict
from enum import Enum


class ScrapingStatus(Enum):
    SUCCESS = "success"
    FAILED = "failed"
    RATE_LIMITED = "rate_limited"
    PARTIAL = "partial"


@dataclass
class RateLimitConfig:
    """Token bucket configuration."""
    requests_per_second: float = 2.0
    burst_size: int = 5
    retry_after_header: str = "Retry-After"


@dataclass
class RetryConfig:
    """Exponential backoff configuration."""
    max_attempts: int = 5
    base_delay: float = 1.0
    max_delay: float = 60.0
    exponential_base: float = 2.0
    jitter: bool = True
    retryable_status_codes: tuple = (429, 500, 502, 503, 504)


@dataclass
class PaginationConfig:
    """Pagination strategy configuration."""
    page_param: str = "page"
    per_page_param: str = "per_page"
    per_page: int = 100
    max_pages: Optional[int] = None
    # For cursor-based pagination
    cursor_param: Optional[str] = None
    next_cursor_path: Optional[str] = None  # JSONPath-like: "meta.next_cursor"


@dataclass
class ScrapingResult:
    """Container for scraping results."""
    url: str
    status: ScrapingStatus
    total_pages: int = 0
    total_records: int = 0
    errors: List[str] = field(default_factory=list)
    started_at: datetime = field(default_factory=datetime.utcnow)
    completed_at: Optional[datetime] = None
    duration_seconds: float = 0.0


# --- Domain Models ---

@dataclass
class User:
    id: int
    username: str
    email: str
    created_at: datetime
    is_active: bool = True
    full_name: Optional[str] = None
    avatar_url: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

    def __post_init__(self):
        if isinstance(self.created_at, str):
            self.created_at = datetime.fromisoformat(
                self.created_at.replace("Z", "+00:00")
            )


@dataclass
class Post:
    id: int
    title: str
    body: str
    user_id: int
    created_at: datetime
    tags: List[str] = field(default_factory=list)
    likes: int = 0
    is_published: bool = False

    def __post_init__(self):
        if isinstance(self.created_at, str):
            self.created_at = datetime.fromisoformat(
                self.created_at.replace("Z", "+00:00")
            )
        if isinstance(self.tags, str):
            self.tags = [t.strip() for t in self.tags.split(",") if t.strip()]


@dataclass
class Comment:
    id: int
    post_id: int
    user_id: int
    body: str
    created_at: datetime
    parent_id: Optional[int] = None

    def __post_init__(self):
        if isinstance(self.created_at, str):
            self.created_at = datetime.fromisoformat(
                self.created_at.replace("Z", "+00:00")
            )

---

rate_limiter.py — Token Bucket Implementation

import asyncio
import time
import logging
from typing import Optional

logger = logging.getLogger(__name__)


class AsyncTokenBucket:
    """
    Thread-safe async token bucket rate limiter.
    
    Supports burst capacity and dynamic rate adjustment
    based on server Retry-After headers.
    """

    def __init__(
        self,
        rate: float,
        burst: int,
        domain: str = "default"
    ):
        self.rate = rate          # tokens per second
        self.burst = burst        # max burst size
        self.domain = domain
        self._tokens = float(burst)
        self._last_refill = time.monotonic()
        self._lock = asyncio.Lock()
        self._backpressure_until: float = 0.0

    def _refill(self) -> None:
        """Refill tokens based on elapsed time."""
        now = time.monotonic()
        elapsed = now - self._last_refill
        tokens_to_add = elapsed * self.rate
        self._tokens = min(self.burst, self._tokens + tokens_to_add)
        self._last_refill = now

    async def acquire(self, tokens: int = 1) -> float:
        """
        Acquire tokens, waiting if necessary.
        Returns the wait time in seconds.
        """
        async with self._lock:
            # Honor server-side backpressure
            now = time.monotonic()
            if now < self._backpressure_until:
                wait = self._backpressure_until - now
                logger.debug(
                    f"[{self.domain}] Backpressure: waiting {wait:.2f}s"
                )
                await asyncio.sleep(wait)

            self._refill()

            if self._tokens >= tokens:
                self._tokens -= tokens
                return 0.0

            # Calculate wait time for enough tokens
            deficit = tokens - self._tokens
            wait_time = deficit / self.rate
            logger.debug(
                f"[{self.domain}] Rate limit hit, waiting {wait_time:.2f}s "
                f"(tokens: {self._tokens:.2f})"
            )
            await asyncio.sleep(wait_time)
            self._refill()
            self._tokens -= tokens
            return wait_time

    async def notify_rate_limited(self, retry_after: Optional[float] = None) -> None:
        """Called when server returns 429 to apply backpressure."""
        async with self._lock:
            if retry_after:
                self._backpressure_until = time.monotonic() + retry_after
                logger.warning(
                    f"[{self.domain}] Server rate limited us. "
                    f"Backing off for {retry_after:.1f}s"
                )
            # Also drain tokens to slow down
            self._tokens = 0

    def get_stats(self) -> dict:
        return {
            "domain": self.domain,
            "rate": self.rate,
            "burst": self.burst,
            "current_tokens": round(self._tokens, 2),
        }


class RateLimiterRegistry:
    """Manages per-domain rate limiters."""

    def __init__(self):
        self._limiters: dict[str, AsyncTokenBucket] = {}
        self._default_rate = 2.0
        self._default_burst = 5

    def get_or_create(
        self,
        domain: str,
        rate: Optional[float] = None,
        burst: Optional[int] = None
    ) -> AsyncTokenBucket:
        if domain not in self._limiters:
            self._limiters[domain] = AsyncTokenBucket(
                rate=rate or self._default_rate,
                burst=burst or self._default_burst,
                domain=domain
            )
        return self._limiters[domain]

    def get_all_stats(self) -> list[dict]:
        return [limiter.get_stats() for limiter in self._limiters.values()]

---

retry.py — Exponential Backoff

import asyncio
import random
import logging
import functools
from typing import TypeVar, Callable, Optional, Tuple, Type
from dataclasses import dataclass

logger = logging.getLogger(__name__)

T = TypeVar("T")


@dataclass
class RetryState:
    attempt: int = 0
    total_delay: float = 0.0
    last_exception: Optional[Exception] = None


class RetryExhaustedError(Exception):
    """Raised when all retry attempts are exhausted."""
    
    def __init__(self, message: str, state: RetryState):
        super().__init__(message)
        self.state = state


def calculate_backoff(
    attempt: int,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    exponential_base: float = 2.0,
    jitter: bool = True
) -> float:
    """
    Calculate backoff delay with optional jitter.
    
    Formula: min(base * exp_base^attempt, max_delay) + jitter
    """
    delay = min(base_delay * (exponential_base ** attempt), max_delay)
    
    if jitter:
        # Full jitter: random between 0 and calculated delay
        # Prevents thundering herd problem
        delay = random.uniform(0, delay)
    
    return delay


class AsyncRetry:
    """
    Async retry decorator with exponential backoff.
    
    Usage:
        @AsyncRetry(max_attempts=3, retryable_exceptions=(aiohttp.ClientError,))
        async def fetch_data():
            ...
    """

    def __init__(
        self,
        max_attempts: int = 5,
        base_delay: float = 1.0,
        max_delay: float = 60.0,
        exponential_base: float = 2.0,
        jitter: bool = True,
        retryable_exceptions: Tuple[Type[Exception], ...] = (Exception,),
        on_retry: Optional[Callable] = None,
    ):
        self.max_attempts = max_attempts
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.exponential_base = exponential_base
        self.jitter = jitter
        self.retryable_exceptions = retryable_exceptions
        self.on_retry = on_retry

    def __call__(self, func: Callable) -> Callable:
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            return await self.execute(func, *args, **kwargs)
        return wrapper

    async def execute(
        self,
        func: Callable,
        *args,
        **kwargs
    ):
        state = RetryState()

        for attempt in range(self.max_attempts):
            state.attempt = attempt
            try:
                result = await func(*args, **kwargs)
                if attempt > 0:
                    logger.info(
                        f"[{func.__name__}] Succeeded after {attempt + 1} attempts "
                        f"(total delay: {state.total_delay:.2f}s)"
                    )
                return result

            except self.retryable_exceptions as e:
                state.last_exception = e
                
                if attempt == self.max_attempts - 1:
                    raise RetryExhaustedError(
                        f"[{func.__name__}] Failed after {self.max_attempts} attempts. "
                        f"Last error: {e}",
                        state=state
                    ) from e

                delay = calculate_backoff(
                    attempt=attempt,
                    base_delay=self.base_delay,
                    max_delay=self.max_delay,
                    exponential_base=self.exponential_base,
                    jitter=self.jitter,
                )
                state.total_delay += delay

                logger.warning(
                    f"[{func.__name__}] Attempt {attempt + 1}/{self.max_attempts} "
                    f"failed: {type(e).__name__}: {e}. "
                    f"Retrying in {delay:.2f}s..."
                )

                if self.on_retry:
                    await self.on_retry(attempt, delay, e)

                await asyncio.sleep(delay)

        raise RetryExhaustedError(
            f"Exhausted {self.max_attempts} attempts",
            state=state
        )

---

extractors.py — Data Extraction & Parsing

import logging
from typing import Any, Dict, List, Optional, Type, TypeVar, get_type_hints
from dataclasses import fields, MISSING
import dataclasses

logger = logging.getLogger(__name__)

T = TypeVar("T")


class ExtractionError(Exception):
    """Raised when data extraction fails."""
    pass


def get_nested(data: Dict, path: str, default: Any = None) -> Any:
    """
    Get nested dict value using dot notation.
    
    Example: get_nested(data, "meta.pagination.next_cursor")
    """
    keys = path.split(".")
    current = data
    for key in keys:
        if isinstance(current, dict):
            current = current.get(key, default)
        elif isinstance(current, list) and key.isdigit():
            idx = int(key)
            current = current[idx] if idx < len(current) else default
        else:
            return default
    return current


def coerce_value(value: Any, target_type: Type) -> Any:
    """Attempt to coerce a value to the target type."""
    if value is None:
        return None
    
    # Handle Optional types
    origin = getattr(target_type, "__origin__", None)
    if origin is type(None):
        return None
    
    # Unwrap Optional[X] -> X
    if origin is type(None) or str(origin) == "typing.Union":
        args = getattr(target_type, "__args__", ())
        non_none_args = [a for a in args if a is not type(None)]
        if non_none_args:
            target_type = non_none_args[0]
    
    try:
        if target_type == bool and isinstance(value, str):
            return value.lower() in ("true", "1", "yes")
        return target_type(value)
    except (ValueError, TypeError):
        return value


def from_dict(cls: Type[T], data: Dict[str, Any]) -> T:
    """
    Safely construct a dataclass from a dictionary,
    handling missing fields with defaults and type coercion.
    """
    if not dataclasses.is_dataclass(cls):
        raise TypeError(f"{cls} is not a dataclass")

    field_map = {}
    type_hints = get_type_hints(cls)

    for f in fields(cls):
        value = data.get(f.name)

        if value is None:
            if f.default is not MISSING:
                field_map[f.name] = f.default
            elif f.default_factory is not MISSING:  # type: ignore
                field_map[f.name] = f.default_factory()  # type: ignore
            else:
                logger.warning(f"Missing required field '{f.name}' for {cls.__name__}")
                field_map[f.name] = None
        else:
            target_type = type_hints.get(f.name, type(value))
            field_map[f.name] = coerce_value(value, target_type)

    return cls(**field_map)


class APIResponseExtractor:
    """
    Extracts structured data from paginated API responses.
    Supports multiple API response formats.
    """

    # Common response wrapper patterns
    DATA_PATHS = ["data", "results", "items", "records", "content"]
    META_PATHS = ["meta", "pagination", "_meta"]

    def extract_items(
        self,
        response: Dict[str, Any],
        data_path: Optional[str] = None
    ) -> List[Dict]:
        """Extract list of items from API response."""
        if data_path:
            items = get_nested(response, data_path)
        else:
            # Auto-detect data location
            items = self._auto_detect_items(response)

        if items is None:
            raise ExtractionError(
                f"Could not find items in response. "
                f"Keys: {list(response.keys())}"
            )

        if not isinstance(items, list):
            raise ExtractionError(
                f"Expected list of items, got {type(items).__name__}"
            )

        return items

    def _auto_detect_items(self, response: Dict) -> Optional[List]:
        """Try common paths to find list data."""
        # Direct list response
        if isinstance(response, list):
            return response
        
        # Try common wrapper keys
        for key in self.DATA_PATHS:
            value = response.get(key)
            if isinstance(value, list):
                return value
        
        return None

    def extract_pagination_info(
        self,
        response: Dict[str, Any],
        cursor_path: Optional[str] = None
    ) -> Dict[str, Any]:
        """Extract pagination metadata from response."""
        info = {
            "total": None,
            "page": None,
            "per_page": None,
            "total_pages": None,
            "has_more": False,
            "next_cursor": None,
            "next_url": None,
        }

        # Try to find pagination data in common locations
        meta = None
        for path in self.META_PATHS:
            meta = response.get(path)
            if meta and isinstance(meta, dict):
                break

        if meta:
            info["total"] = meta.get("total") or meta.get("total_count")
            info["page"] = meta.get("page") or meta.get("current_page")
            info["per_page"] = meta.get("per_page") or meta.get("limit")
            info["total_pages"] = meta.get("total_pages") or meta.get("pages")
            info["has_more"] = bool(meta.get("has_more") or meta.get("has_next"))
            info["next_url"] = meta.get("next") or meta.get("next_url")
            info["next_cursor"] = meta.get("cursor") or meta.get("next_cursor")

        # Cursor-based pagination via custom path
        if cursor_path:
            info["next_cursor"] = get_nested(response, cursor_path)

        # Infer has_more
        if info["page"] and info["total_pages"]:
            info["has_more"] = info["page"] < info["total_pages"]

        return info

    def parse_items(
        self,
        items: List[Dict],
        model_cls: Type[T],
        strict: bool = False
    ) -> Tuple[List[T], List[Dict]]:
        """
        Parse raw dicts into typed dataclass instances.
        
        Returns (successful_items, failed_items)
        """
        from typing import Tuple  # local import to avoid circular
        successful = []
        failed = []

        for item in items:
            try:
                instance = from_dict(model_cls, item)
                successful.append(instance)
            except Exception as e:
                logger.error(f"Failed to parse {model_cls.__name__}: {e} | Data: {item}")
                failed.append({"data": item, "error": str(e)})
                if strict:
                    raise ExtractionError(f"Strict mode: {e}") from e

        logger.debug(
            f"Parsed {len(successful)}/{len(items)} {model_cls.__name__} records "
            f"({len(failed)} failures)"
        )
        return successful, failed

---

scraper.py — Core Async Scraper

import asyncio
import logging
import time
from dataclasses import dataclass, field
from datetime import datetime
from typing import (
    AsyncIterator, Dict, Generic, List, Optional,
    Tuple, Type, TypeVar, Any
)
from urllib.parse import urlencode, urlparse

import aiohttp
from aiohttp import ClientResponseError, ClientConnectorError, ServerTimeoutError

from models import (
    RateLimitConfig, RetryConfig, PaginationConfig,
    ScrapingResult, ScrapingStatus
)
from rate_limiter import AsyncTokenBucket, RateLimiterRegistry
from retry import AsyncRetry, RetryExhaustedError, calculate_backoff
from extractors import APIResponseExtractor, ExtractionError

logger = logging.getLogger(__name__)

T = TypeVar("T")

# Exceptions we'll retry on
RETRYABLE_EXCEPTIONS = (
    ClientConnectorError,
    ServerTimeoutError,
    asyncio.TimeoutError,
    ConnectionResetError,
)


@dataclass
class FetchResult:
    """Raw result from a single HTTP request."""
    url: str
    status_code: int
    data: Optional[Dict] = None
    error: Optional[str] = None
    headers: Dict[str, str] = field(default_factory=dict)
    duration_ms: float = 0.0
    attempt: int = 1


class AsyncScraper(Generic[T]):
    """
    Production-grade async web scraper with:
    - Per-domain rate limiting (token bucket)
    - Exponential backoff retries
    - Automatic pagination handling
    - Typed data extraction
    - Connection pooling
    - Comprehensive error handling
    """

    def __init__(
        self,
        base_url: str,
        model_cls: Type[T],
        rate_config: Optional[RateLimitConfig] = None,
        retry_config: Optional[RetryConfig] = None,
        pagination_config: Optional[PaginationConfig] = None,
        default_headers: Optional[Dict[str, str]] = None,
        timeout_seconds: float = 30.0,
        max_concurrent_requests: int = 10,
    ):
        self.base_url = base_url.rstrip("/")
        self.model_cls = model_cls
        self.rate_config = rate_config or RateLimitConfig()
        self.retry_config = retry_config or RetryConfig()
        self.pagination_config = pagination_config or PaginationConfig()
        self.default_headers = default_headers or {}
        self.timeout_seconds = timeout_seconds

        # Domain for rate limiting
        self._domain = urlparse(base_url).netloc

        # Session management
        self._session: Optional[aiohttp.ClientSession] = None
        self._connector: Optional[aiohttp.TCPConnector] = None

        # Rate limiter
        self._limiter = AsyncTokenBucket(
            rate=self.rate_config.requests_per_second,
            burst=self.rate_config.burst_size,
            domain=self._domain,
        )

        # Semaphore for concurrency control
        self._semaphore = asyncio.Semaphore(max_concurrent_requests)

        # Data extractor
        self._extractor = APIResponseExtractor()

        # Stats
        self._total_requests = 0
        self._failed_requests = 0
        self._rate_limited_count = 0

    async def __aenter__(self) -> "AsyncScraper":
        await self._create_session()
        return self

    async def __aexit__(self, *args) -> None:
        await self._close_session()

    async def _create_session(self) -> None:
        """Create aiohttp session with connection pooling."""
        self._connector = aiohttp.TCPConnector(
            limit=100,           # Total connection pool size
            limit_per_host=20,   # Per-host connection limit
            ttl_dns_cache=300,   # DNS cache TTL
            enable_cleanup_closed=True,
        )
        self._session = aiohttp.ClientSession(
            connector=self._connector,
            headers={
                "User-Agent": "AsyncScraper/1.0",
                "Accept": "application/json",
                **self.default_headers,
            },
            timeout=aiohttp.ClientTimeout(total=self.timeout_seconds),
        )
        logger.info(f"Session created for domain: {self._domain}")

    async def _close_session(self) -> None:
        """Gracefully close the session."""
        if self._session:
            await self._session.close()
            # Allow SSL cleanup
            await asyncio.sleep(0.25)
        logger.info(
            f"Session closed. Stats — Requests: {self._total_requests}, "
            f"Failed: {self._failed_requests}, "
            f"Rate limited: {self._rate_limited_count}"
        )

    async def _fetch_with_retry(
        self,
        url: str,
        params: Optional[Dict] = None,
        attempt: int = 0
    ) -> FetchResult:
        """
        Fetch a URL with rate limiting and exponential backoff.
        Recursive retry implementation.
        """
        if attempt >= self.retry_config.max_attempts:
            raise RetryExhaustedError(
                f"Max retries ({self.retry_config.max_attempts}) exhausted for {url}",
                state=None
            )

        # Acquire rate limit token
        await self._limiter.acquire()

        async with self._semaphore:
            start_time = time.monotonic()
            self._total_requests += 1

            try:
                async with self._session.get(url, params=params) as response:
                    duration_ms = (time.monotonic() - start_time) * 1000
                    
                    logger.debug(
                        f"GET {url} | Status: {response.status} | "
                        f"Duration: {duration_ms:.0f}ms | Attempt: {attempt + 1}"
                    )

                    # Handle rate limiting (429)
                    if response.status == 429:
                        self._rate_limited_count += 1
                        retry_after = self._parse_retry_after(response)
                        await self._limiter.notify_rate_limited(retry_after)
                        
                        delay = retry_after or calculate_backoff(
                            attempt,
                            self.retry_config.base_delay,
                            self.retry_config.max_delay,
                            jitter=self.retry_config.jitter
                        )
                        logger.warning(
                            f"Rate limited (429) on {url}. "
                            f"Retrying after {delay:.1f}s..."
                        )
                        await asyncio.sleep(delay)
                        return await self._fetch_with_retry(url, params, attempt + 1)

                    # Handle other retryable status codes
                    if response.status in self.retry_config.retryable_status_codes:
                        self._failed_requests += 1
                        delay = calculate_backoff(
                            attempt,
                            self.retry_config.base_delay,
                            self.retry_config.max_delay,
                            jitter=self.retry_config.jitter
                        )
                        logger.warning(
                            f"Retryable status {response.status} on {url}. "
                            f"Attempt {attempt + 1}. Retrying in {delay:.1f}s..."
                        )
                        await asyncio.sleep(delay)
                        return await self._fetch_with_retry(url, params, attempt + 1)

                    # Non-retryable errors
                    if response.status >= 400:
                        error_text = await response.text()
                        return FetchResult(
                            url=url,
                            status_code=response.status,
                            error=f"HTTP {response.status}: {error_text[:200]}",
                            duration_ms=duration_ms,
                            attempt=attempt + 1,
                        )

                    # Success
                    data = await response.json(content_type=None)
                    return FetchResult(
                        url=url,
                        status_code=response.status,
                        data=data,
                        headers=dict(response.headers),
                        duration_ms=duration_ms,
                        attempt=attempt + 1,
                    )

            except RETRYABLE_EXCEPTIONS as e:
                self._failed_requests += 1
                delay = calculate_backoff(
                    attempt,
                    self.retry_config.base_delay,
                    self.retry_config.max_delay,
                    jitter=self.retry_config.jitter
                )
                logger.warning(
                    f"Connection error on {url}: {type(e).__name__}: {e}. "
                    f"Retrying in {delay:.1f}s..."
                )
                await asyncio.sleep(delay)
                return await self._fetch_with_retry(url, params, attempt + 1)

    def _parse_retry_after(self, response: aiohttp.ClientResponse) -> Optional[float]:
        """Parse Retry-After header (seconds or HTTP date)."""
        header = response.headers.get(self.rate_config.retry_after_header)
        if not header:
            return None
        try:
            return float(header)
        except ValueError:
            # Try parsing as HTTP date
            from email.utils import parsedate
            parsed = parsedate(header)
            if parsed:
                import calendar
                target = calendar.timegm(parsed)
                return max(0, target - time.time())
        return None

    def _build_url(self, endpoint: str) -> str:
        """Build full URL from base and endpoint."""
        if endpoint.startswith("http"):
            return endpoint
        return f"{self.base_url}/{endpoint.lstrip('/')}"

    async def scrape_page(
        self,
        endpoint: str,
        page: int = 1,
        extra_params: Optional[Dict] = None,
        data_path: Optional[str] = None,
    ) -> Tuple[List[T], Dict]:
        """
        Scrape a single page and return parsed items + pagination info.
        """
        url = self._build_url(endpoint)
        params = {
            self.pagination_config.page_param: page,
            self.pagination_config.per_page_param: self.pagination_config.per_page,
            **(extra_params or {}),
        }

        result = await self._fetch_with_retry(url, params=params)

        if result.error or not result.data:
            logger.error(f"Failed to fetch page {page}: {result.error}")
            return [], {}

        items_raw = self._extractor.extract_items(result.data, data_path)
        pagination = self._extractor.extract_pagination_info(
            result.data,
            cursor_path=self.pagination_config.next_cursor_path,
        )
        parsed_items, failed = self._extractor.parse_items(items_raw, self.model_cls)

        if failed:
            logger.warning(f"Page {page}: {len(failed)} items failed to parse")

        return parsed_items, pagination

    async def scrape_all_pages(
        self,
        endpoint: str,
        extra_params: Optional[Dict] = None,
        data_path: Optional[str] = None,
    ) -> AsyncIterator[List[T]]:
        """
        Async generator that yields items page by page.
        
        Usage:
            async for page_items in scraper.scrape_all_pages("/users"):
                await save_to_db(page_items)
        """
        page = 1
        total_records = 0
        max_pages = self.pagination_config.max_pages

        logger.info(f"Starting pagination scrape: {endpoint}")

        while True:
            logger.info(f"Fetching page {page}...")

            items, pagination = await self.scrape_page(
                endpoint, page, extra_params, data_path
            )

            if not items:
                logger.info(f"No items on page {page}, stopping.")
                break

            total_records += len(items)
            yield items

            logger.info(
                f"Page {page}: {len(items)} items "
                f"(total so far: {total_records}) | "
                f"Pagination: {pagination}"
            )

            # Determine if there

Here's a production-ready, asynchronous Python web scraper that handles rate limiting, retries with exponential backoff, paginated API responses, and extracts data into typed dataclasses. It's built using aiohttp, asyncio, and Python's dataclasses.

import asyncio
import aiohttp
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
from datetime import datetime
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


@dataclass
class Item:
    """Represents a single data item from the API."""
    id: int
    title: str
    description: Optional[str] = None
    created_at: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'Item':
        return cls(
            id=data.get('id'),
            title=data.get('title'),
            description=data.get('description'),
            created_at=data.get('created_at'),
            metadata=data.get('metadata', {})
        )


@dataclass
class ScrapedResult:
    """Container for all scraped items and metadata."""
    items: List[Item] = field(default_factory=list)
    total_pages: int = 0
    total_items: int = 0
    scrape_duration: float = 0.0
    errors: List[Dict[str, Any]] = field(default_factory=list)
    timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())


class RateLimiter:
    """Token bucket rate limiter for async operations."""

    def __init__(self, rate: int, per: float = 1.0):
        self.rate = rate
        self.per = per
        self.tokens = rate
        self.last_update = asyncio.get_event_loop().time()
        self._lock = asyncio.Lock()

    async def acquire(self):
        async with self._lock:
            now = asyncio.get_event_loop().time()
            elapsed = now - self.last_update
            self.tokens = min(self.rate, self.tokens + elapsed * (self.rate / self.per))
            self.last_update = now

            if self.tokens < 1:
                wait_time = (1 - self.tokens) * (self.per / self.rate)
                await asyncio.sleep(wait_time)
                self.tokens = 0
            else:
                self.tokens -= 1


class AsyncScraper:
    """Async web scraper with rate limiting, retries, and pagination."""

    def __init__(
        self,
        base_url: str,
        rate_limit: int = 10,
        rate_limit_per: float = 1.0,
        max_retries: int = 3,
        timeout: float = 30.0,
        concurrency: int = 5
    ):
        self.base_url = base_url
        self.rate_limiter = RateLimiter(rate_limit, rate_limit_per)
        self.max_retries = max_retries
        self.timeout = aiohttp.ClientTimeout(total=timeout)
        self.semaphore = asyncio.Semaphore(concurrency)
        self.errors: List[Dict[str, Any]] = []

    async def _fetch_with_retry(
        self,
        session: aiohttp.ClientSession,
        url: str,
        attempt: int = 1
    ) -> Optional[Dict[str, Any]]:
        async with self.semaphore:
            await self.rate_limiter.acquire()
            try:
                async with session.get(url, timeout=self.timeout) as resp:
                    resp.raise_for_status()
                    return await resp.json()
            except aiohttp.ClientError as e:
                wait = (2 ** attempt) + (asyncio.get_event_loop().time() % 1)
                if attempt < self.max_retries:
                    logger.warning(f"Retrying {url} (attempt {attempt + 1}) after {wait:.2f}s: {e}")
                    await asyncio.sleep(wait)
                    return await self._fetch_with_retry(session, url, attempt + 1)
                else:
                    logger.error(f"Failed to fetch {url} after {self.max_retries} retries: {e}")
                    self.errors.append({"url": url, "attempt": attempt, "error": str(e)})
                    return None

    async def _fetch_page(
        self,
        session: aiohttp.ClientSession,
        page: int,
        page_size: int = 20
    ) -> Optional[Dict[str, Any]]:
        url = f"{self.base_url}?page={page}&per_page={page_size}"
        return await self._fetch_with_retry(session, url)

    async def scrape_all_pages(
        self,
        page_size: int = 20,
        max_pages: Optional[int] = None
    ) -> ScrapedResult:
        start_time = asyncio.get_event_loop().time()
        items: List[Item] = []
        page = 1
        total_pages = 0

        async with aiohttp.ClientSession() as session:
            while True:
                logger.info(f"Fetching page {page}")
                data = await self._fetch_page(session, page, page_size)
                if not data:
                    break

                total_pages = data.get('total_pages', 1)
                page_items = data.get('data', [])
                items.extend(Item.from_dict(item) for item in page_items)

                if max_pages and page >= max_pages:
                    break
                if page >= total_pages:
                    break
                page += 1

        duration = asyncio.get_event_loop().time() - start_time
        return ScrapedResult(
            items=items,
            total_pages=total_pages,
            total_items=len(items),
            scrape_duration=duration,
            errors=self.errors
        )


# Example usage
async def main():
    scraper = AsyncScraper(
        base_url="https://api.example.com/items",
        rate_limit=5,
        rate_limit_per=1.0,
        max_retries=3,
        timeout=30.0,
        concurrency=5
    )
    result = await scraper.scrape_all_pages(page_size=20, max_pages=10)

    print(f"Scraped {result.total_items} items from {result.total_pages} pages")
    print(f"Duration: {result.scrape_duration:.2f}s")
    if result.errors:
        print(f"Errors encountered: {len(result.errors)}")
        for err in result.errors:
            print(err)


if __name__ == "__main__":
    asyncio.run(main())

Key Features

• Async HTTP with aiohttp: Efficient concurrent requests.
• Rate Limiting: Token bucket algorithm ensures controlled request rates.
• Exponential Backoff Retries: Resilient to transient network/API issues.
• Pagination Support: Automatically iterates through paginated API results.
• Typed Dataclasses: Ensures structured, typed output for downstream use.
• Error Tracking: Collects and reports failed requests for audit/debug.
• Configurable Concurrency & Timeout: Tunable for various API constraints.

Installation

pip install aiohttp

This modular scraper can be extended with authentication, caching, or data export as needed for your specific use case.