Plugin System Overview

The scraping framework uses a decorator-based plugin system to keep the pipeline extensible without modifying core code.

Architecture

graph TD
    Registry["core/registry.py<br/>(decorator registry)"]
    Fetcher["BaseFetcherPlugin"]
    Parser["BaseParserPlugin"]
    Storage["BaseStoragePlugin"]

    subgraph Fetchers
        CB[cloakbrowser_fetcher]
        WR[wreq_fetcher]
    end
    subgraph Parsers
        GP[generic_parser]
        JP[jsonld_parser]
        AP[apollo_json_parser]
        G2[g2_listing_parser]
        HP[hermes_listing_parser]
        IP[idealista_detail]
        TP[trustpilot_nextdata_parser]
        JPP[jsonpath_parser]
    end
    subgraph Storage
        S3[garage_s3_storage]
        PG[postgres_storage]
        JS[json_storage]
        CS[csv_storage]
        MS[markdown_storage]
    end

    Fetcher --> Registry
    Parser --> Registry
    Storage --> Registry

    Registry --> CB & WR
    Registry --> GP & JP & AP & G2 & HP & IP & TP & JPP
    Registry --> S3 & PG & JS & CS & MS

Plugin Registration

Plugins register themselves via decorators:

from core.registry import register_fetcher, register_parser, register_storage

@register_fetcher
class MyFetcher(BaseFetcherPlugin):
    plugin_id = "my_fetcher"

    async def fetch(self, url: str, context: dict) -> dict:
        ...

The registry is populated by registry._load_builtins() which is called once at worker startup. Profiles reference plugins by their plugin_id:

# profile.yaml
stages:
  listing:
    plugin: my_fetcher     # ← plugin_id
    parser: generic         # ← parser plugin_id

Base Classes

All plugins inherit from abstract base classes in plugins/base.py:

BaseFetcherPlugin

class BaseFetcherPlugin(ABC):
    plugin_id: str

    @abstractmethod
    async def fetch(self, url: str, context: dict) -> dict:
        """Fetch a URL and return raw response data.

        Returns:
            {"url": str, "status": int, "html": str, "headers": dict}
        """

BaseParserPlugin

class BaseParserPlugin(ABC):
    plugin_id: str

    @classmethod
    def from_stage_config(cls, selectors, stage_config):
        """Build parser from stage configuration."""

    @abstractmethod
    def parse(self, raw: dict) -> list[dict]:
        """Parse raw HTML into structured items.

        Args:
            raw: {"url": str, "status": int, "html": str, "headers": dict, "plugin_id": str}
        Returns:
            list of structured item dicts
        """

BaseStoragePlugin

class BaseStoragePlugin(ABC):
    plugin_id: str

    @abstractmethod
    async def store(self, items: list[dict]) -> None:
        """Persist a batch of structured items."""

---

[!TIP] Use Claude to enrich this section with detailed documentation for each plugin's internal logic, edge cases, and configuration options.