Authoring Plugins

Ladon crawl plugins live in separate repos — the core library is intentionally unaware of any site-specific logic. This keeps the framework dependency-free and each adapter independently versioned.

Ethical crawling: enable robots.txt enforcement

When writing a plugin that targets public third-party websites, configure your HttpClient with respect_robots_txt=True. This is the IETF standard (RFC 9309), the established industry norm, and increasingly a legal expectation under EU data-protection law. See Getting Started → Ethical note for the full rationale.

The plugin protocol

A plugin must implement the CrawlPlugin protocol:

from ladon.plugins.protocol import CrawlPlugin, Expander, Sink, Source

class MyPlugin:
    name: str                  # short identifier used in logs
    source: Source             # top-level ref discovery
    expanders: list[Expander]  # ordered chain of URL/ref expanders
    sink: Sink                 # leaf processor

Ladon uses structural subtyping (PEP 544 Protocol). No inheritance is required — your class just needs to provide the attributes above. Instance attributes set in __init__ satisfy the protocol at runtime, so the common pattern is:

from ladon import SyncHttpClientProtocol

class MyPlugin:
    def __init__(self, client: SyncHttpClientProtocol) -> None:
        self.name = "my_plugin"
        self.source = MySource()
        self.expanders = [MyExpander()]
        self.sink = MySink()

CLI constructor requirement

When invoked via ladon run --plugin, the CLI constructs your plugin as plugin_cls(client=client). Make sure your __init__ accepts client as a keyword argument.

Expander

An Expander turns one ref into an Expansion — the current node's record plus the child refs to process next (e.g. catalog record + product URLs):

from ladon import SyncHttpClientProtocol
from ladon.plugins.models import Expansion

class MyExpander:
    def expand(
        self, ref: object, client: SyncHttpClientProtocol
    ) -> Expansion:
        """Fetch ref; return its record and child refs.

        Raises:
            ExpansionNotReadyError: ref is not yet ready to be expanded.
            PartialExpansionError: child list is incomplete.
            ChildListUnavailableError: child list could not be retrieved.
        """
        ...

Expansion exception handling depends on where the failure occurs:

Exception	Meaning
ExpansionNotReadyError	Data not ready; abort the entire run — caller retries later
PartialExpansionError	Some children unavailable; abort from the first Expander, otherwise record and skip only the failing branch
ChildListUnavailableError	Child list fetch failed; abort from the first Expander, otherwise record and skip only the failing branch

Sink

A Sink processes each leaf ref (e.g. downloads a product page):

from ladon import SyncHttpClientProtocol

class MySink:
    def consume(
        self, ref: object, client: SyncHttpClientProtocol
    ) -> object:
        """Fetch and process the leaf; return a record for on_leaf callback."""
        ...

LeafUnavailableError signals that the leaf is temporarily unavailable; the runner records the failure and moves on. Unexpected exceptions from a Sink are recorded the same way so other independent leaves still run; their text remains visible in RunResult.errors. Cancellation and explicitly fatal AssetDownloadError propagate instead of being counted as leaf failures. Expansion exceptions also retain their typed contract when raised by a Sink: ExpansionNotReadyError, PartialExpansionError, and ChildListUnavailableError propagate unchanged.

CrawlPlugin

Combine expanders and sink into a plugin:

from ladon import SyncHttpClientProtocol

class ShopPlugin:
    def __init__(self, client: SyncHttpClientProtocol) -> None:
        self.name = "shop_example"
        self.source = CatalogSource()
        self.expanders = [CategoryExpander(), ProductExpander()]
        self.sink = ProductSink()

Running from code

from ladon.networking.client import HttpClient
from ladon.networking.config import HttpClientConfig
from ladon.runner import RunConfig, run_plugin

config = HttpClientConfig(retries=2, min_request_interval_seconds=1.0)
client = HttpClient(config)
plugin = ShopPlugin(client=client)

result = run_plugin(
    plugin=plugin,
    client=client,
    config=RunConfig(leaf_limit=100),
    on_leaf=lambda leaf_record, parent_record: db.save(leaf_record),
)
print(f"fetched {result.leaves_consumed}, failed {result.leaves_failed}")
client.close()

run_plugin() calls plugin.source.discover(client) once and processes each returned root in source order. Its PluginRunResult.results preserves each root's RunResult; leaf_limit applies per root. Use run_crawl(top_ref, ...) when your application deliberately owns discovery or has one known root. If a later root raises a globally fatal error, earlier roots may already have called on_leaf; make persistence callbacks idempotent before retrying a whole-plugin run.

Running from the CLI

ladon run --plugin mypackage.adapters:ShopPlugin \
          --ref https://example-shop.com/categories/electronics

The CLI uses default RunConfig settings (no leaf limit, no on_leaf callback). For production use write a Python script that usually calls run_plugin(); use run_crawl() when you need explicit per-root dispatch.

Async plugins

For high-concurrency crawls implement AsyncCrawlPlugin and call async_run_plugin() instead. The async protocols mirror the sync ones with async def methods and AsyncHttpClientProtocol as the client parameter.

from ladon import AsyncHttpClientProtocol
from ladon.plugins.async_protocol import AsyncCrawlPlugin, AsyncExpander, AsyncSink, AsyncSource

class AsyncShopPlugin:
    def __init__(self) -> None:
        self.name = "async_shop_example"
        self.source = AsyncCatalogSource()
        self.expanders = [AsyncCategoryExpander(), AsyncProductExpander()]
        self.sink = AsyncProductSink()


class AsyncProductSink:
    async def consume(
        self, ref: object, client: AsyncHttpClientProtocol
    ) -> object:
        result = await client.get(str(ref))
        if not result.ok:
            from ladon.plugins.errors import LeafUnavailableError
            raise LeafUnavailableError(f"fetch failed: {result.error}")
        return parse_product(result.value)

Run it:

import asyncio
from ladon import AsyncHttpClient, async_run_plugin
from ladon.networking.config import HttpClientConfig
from ladon.runner import RunConfig

async def main() -> None:
    config = HttpClientConfig(retries=2, min_request_interval_seconds=0.5)
    async with AsyncHttpClient(config) as client:
        result = await async_run_plugin(
            plugin=AsyncShopPlugin(),
            client=client,
            config=RunConfig(leaf_limit=100, async_concurrency=20),
            on_leaf=my_async_persist,
        )
        print(f"fetched {result.leaves_consumed}, failed {result.leaves_failed}")

asyncio.run(main())

async_run_plugin accepts the same RunConfig as run_plugin. The async_concurrency field controls how many leaf fetches run simultaneously (within each discovered root; default 10). The sync runner ignores it.

Error taxonomy

All errors are in ladon.plugins.errors and ladon.networking.errors.

Error	Layer	Meaning
ExpansionNotReadyError	Plugin	Run not yet possible; abort and retry later
PartialExpansionError	Plugin	Some branches unavailable; log and continue
ChildListUnavailableError	Plugin	Child list fetch failed
LeafUnavailableError	Plugin	Individual leaf unavailable
CircuitOpenError	Networking	Host circuit breaker is open
RobotsBlockedError	Networking	robots.txt disallows the URL
RequestTimeoutError	Networking	Request exceeded timeout
TransientNetworkError	Networking	Connection-level transport failure; all internal retries exhausted