Cookbook

These patterns are small, complete crawl building blocks. They use Ladon's current Source → Expander → Sink contracts: a source discovers root Refs, each expander returns an Expansion, and the sink turns leaf refs into records. run_plugin() drives the source and each discovered root; use the single-root run_crawl() only when your caller deliberately owns dispatch. Replace the parsing and URLs with those for the site you are allowed to crawl.

Flat crawl: one listing page to item records

Use one expander when every page in a paginated listing contains the leaves. Have your Source.discover() return one Ref per page, then call run_plugin() once. This example uses a public GitHub API listing page; the page value is context you can use in logs or persistence.

This example and the Hacker News tree crawl below use live external services, so they are verified by a weekly scheduled GitHub Actions check rather than the offline pytest suite.

import json
from collections.abc import Sequence
from dataclasses import dataclass

from ladon import (
    ChildListUnavailableError,
    Expansion,
    HttpClient,
    HttpClientConfig,
    LeafUnavailableError,
    PluginRunResult,
    Ref,
    RunConfig,
    SyncHttpClientProtocol,
    run_plugin,
)


class GitHubIssuesSource:
    def discover(self, client: SyncHttpClientProtocol) -> Sequence[Ref]:
        return [
            Ref(
                "https://api.github.com/repos/psf/requests/issues?per_page=5"
                f"&page={page}",
                {"page": page},
            )
            for page in (1, 2)
        ]


class IssueList:
    def expand(self, ref: Ref, client: SyncHttpClientProtocol) -> Expansion:
        response = client.get(ref.url)
        if not response.ok or response.value is None:
            raise ChildListUnavailableError(f"listing failed: {response.error}")
        issues = json.loads(response.value)
        return Expansion(
            {"page": ref.raw["page"]}, [Ref(issue["url"]) for issue in issues]
        )


class IssueSink:
    def consume(
        self, ref: Ref, client: SyncHttpClientProtocol
    ) -> dict[str, object]:
        response = client.get(ref.url)
        if not response.ok or response.value is None:
            raise LeafUnavailableError(f"issue failed: {response.error}")
        issue = json.loads(response.value)
        return {"number": issue["number"], "title": issue["title"]}


@dataclass(frozen=True)
class GitHubIssuesPlugin:
    name: str = "github-issues"
    source: GitHubIssuesSource = GitHubIssuesSource()
    expanders: Sequence[IssueList] = (IssueList(),)
    sink: IssueSink = IssueSink()


def run_example() -> PluginRunResult:
    plugin = GitHubIssuesPlugin()
    with HttpClient(
        HttpClientConfig(user_agent="example-crawler/1.0")
    ) as client:
        result = run_plugin(plugin, client, RunConfig())
        print(result.leaves_consumed, result.errors)
        return result


if __name__ == "__main__":
    run_example()

Multi-level tree crawl: Hacker News

The ladon-hackernews adapter is a working example of a front page → story → comment tree. Its HNSource discovers top-story refs, HNExpander fetches one story and emits direct comment refs, and HNSink fetches each comment. Install it with pip install ladon-hackernews before running this example.

from operator import attrgetter

from ladon_hackernews import HNPlugin

from ladon import (
    ExpansionNotReadyError,
    HttpClient,
    HttpClientConfig,
    RunConfig,
    RunResult,
    run_crawl,
)


def print_comment(comment: object, story: object) -> None:
    # Full precision requires ladon-hackernews to narrow its object signatures.
    get_id = attrgetter("id")
    print(get_id(story), get_id(comment))


def run_example() -> list[RunResult]:
    plugin = HNPlugin(top=10)
    results: list[RunResult] = []
    with HttpClient(
        HttpClientConfig(user_agent="my-hn-research-bot/1.0")
    ) as client:
        for story_ref in plugin.source.discover(client):
            try:
                result = run_crawl(
                    story_ref,
                    plugin,
                    client,
                    RunConfig(leaf_limit=50),
                    on_leaf=print_comment,
                )
            except ExpansionNotReadyError:
                continue  # Rediscover this story on the next scheduled run.
            results.append(result)
            print(result.leaves_consumed, result.leaves_failed, result.errors)
    return results


if __name__ == "__main__":
    run_example()

The adapter has one configured expander because the source is outside the runner: HNSource covers the front-page → story edge, and HNExpander covers the story → comment edge.

Carry listing context in ref.raw

When a listing already contains fields the sink needs, put them in the child ref rather than requesting the item page again. Ref is frozen, so construct a new ref with its raw mapping in the expander.

import json
from collections.abc import Sequence
from dataclasses import dataclass
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread
from typing import cast

from ladon import (
    ChildListUnavailableError,
    CrawlPlugin,
    Expansion,
    HttpClient,
    HttpClientConfig,
    LeafUnavailableError,
    Ref,
    RunConfig,
    RunResult,
    SyncHttpClientProtocol,
    run_crawl,
)


class ProductHandler(BaseHTTPRequestHandler):
    def do_GET(self) -> None:  # noqa: N802
        if self.path != "/products":
            self.send_error(404)
            return
        host, port = cast(tuple[str, int], self.server.server_address)
        products = {
            "products": [
                {
                    "url": f"http://{host}:{port}/product/tea",
                    "sku": "TEA-1",
                    "price": 7,
                },
                {
                    "url": f"http://{host}:{port}/product/mug",
                    "sku": "MUG-2",
                    "price": 12,
                },
            ]
        }
        body = json.dumps(products).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format: str, *args: object) -> None:
        pass


def build_mock_server() -> tuple[ThreadingHTTPServer, Thread]:
    server = ThreadingHTTPServer(("127.0.0.1", 0), ProductHandler)
    thread = Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, thread


class ProductList:
    def expand(self, ref: Ref, client: SyncHttpClientProtocol) -> Expansion:
        response = client.get(ref.url)
        if not response.ok or response.value is None:
            raise ChildListUnavailableError(f"listing failed: {response.error}")
        products = json.loads(response.value)["products"]
        children = [
            Ref(
                product["url"],
                raw={"sku": product["sku"], "price": product["price"]},
            )
            for product in products
        ]
        return Expansion(record={"category": ref.url}, child_refs=children)


class ProductSink:
    def consume(
        self, ref: Ref, client: SyncHttpClientProtocol
    ) -> dict[str, object]:
        if "sku" not in ref.raw:
            raise LeafUnavailableError("listing did not provide a SKU")
        return {"sku": ref.raw["sku"], "price": ref.raw["price"]}


class ProductSource:
    def discover(self, client: SyncHttpClientProtocol) -> Sequence[Ref]:
        return ()


@dataclass(frozen=True)
class ProductPlugin:
    name: str = "local-products"
    source: ProductSource = ProductSource()
    expanders: tuple[ProductList, ...] = (ProductList(),)
    sink: ProductSink = ProductSink()


def run_example() -> RunResult:
    server, thread = build_mock_server()
    host, port = cast(tuple[str, int], server.server_address)
    try:
        with HttpClient(
            HttpClientConfig(user_agent="example-crawler/1.0")
        ) as client:
            result = run_crawl(
                Ref(f"http://{host}:{port}/products"),
                cast("CrawlPlugin", ProductPlugin()),
                client,
                RunConfig(),
            )
        print(result.leaves_consumed, result.errors)
        return result
    finally:
        server.shutdown()
        server.server_close()
        thread.join()


if __name__ == "__main__":
    run_example()

This sink deliberately makes no HTTP request: its record is built from data the expander already fetched. The Hacker News adapter uses the same technique to carry story_id from HNExpander to HNSink.

Resume a not-ready or partial crawl

Catch runner-level expansion errors at the scheduling boundary. An ExpansionNotReadyError always propagates: do not retry it in the same run. A PartialExpansionError also propagates when the first expander raises it; at later levels the runner skips only that branch and adds its message to RunResult.errors.

import json
from collections.abc import Sequence
from dataclasses import dataclass
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread
from typing import cast
from urllib.parse import parse_qs, urlparse

from ladon import (
    ChildListUnavailableError,
    CrawlPlugin,
    Expansion,
    ExpansionNotReadyError,
    HttpClient,
    HttpClientConfig,
    LeafUnavailableError,
    PartialExpansionError,
    Ref,
    RunConfig,
    RunResult,
    SyncHttpClientProtocol,
    run_crawl,
)


class CrawlStateHandler(BaseHTTPRequestHandler):
    def do_GET(self) -> None:  # noqa: N802
        parsed = urlparse(self.path)
        query = parse_qs(parsed.query)
        host, port = cast(tuple[str, int], self.server.server_address)
        base_url = f"http://{host}:{port}"
        if parsed.path == "/top":
            mode = query.get("mode", ["complete"])[0]
            body = {"status": mode, "categories": ["tea", "mugs"]}
        elif parsed.path == "/category/tea":
            body = {"status": "complete", "items": [f"{base_url}/item/tea"]}
        elif parsed.path == "/category/mugs":
            body = {"status": "partial", "items": [f"{base_url}/item/mug"]}
        elif parsed.path.startswith("/item/"):
            body = {"name": parsed.path.rsplit("/", 1)[1]}
        else:
            self.send_error(404)
            return
        encoded = json.dumps(body).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(encoded)))
        self.end_headers()
        self.wfile.write(encoded)

    def log_message(self, format: str, *args: object) -> None:
        pass


def build_mock_server() -> tuple[ThreadingHTTPServer, Thread]:
    server = ThreadingHTTPServer(("127.0.0.1", 0), CrawlStateHandler)
    thread = Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, thread


def load_json(ref: Ref, client: SyncHttpClientProtocol) -> dict[str, object]:
    response = client.get(ref.url)
    if not response.ok or response.value is None:
        raise ChildListUnavailableError(f"request failed: {response.error}")
    return json.loads(response.value)


class CategoryList:
    def expand(self, ref: Ref, client: SyncHttpClientProtocol) -> Expansion:
        payload = load_json(ref, client)
        status = cast(str, payload["status"])
        if status == "not-ready":
            raise ExpansionNotReadyError("listing is not published yet")
        if status == "partial":
            raise PartialExpansionError("top-level listing is incomplete")
        return Expansion(
            record={"status": status},
            child_refs=[
                Ref(f"{ref.url.rsplit('/top', 1)[0]}/category/{name}")
                for name in cast(list[str], payload["categories"])
            ],
        )


class ItemList:
    def expand(self, ref: Ref, client: SyncHttpClientProtocol) -> Expansion:
        payload = load_json(ref, client)
        if payload["status"] == "partial":
            raise PartialExpansionError("category still has another page")
        return Expansion(
            record={"category": ref.url},
            child_refs=[Ref(url) for url in cast(list[str], payload["items"])],
        )


class ItemSink:
    def consume(
        self, ref: Ref, client: SyncHttpClientProtocol
    ) -> dict[str, object]:
        response = client.get(ref.url)
        if not response.ok or response.value is None:
            raise LeafUnavailableError(f"item failed: {response.error}")
        return json.loads(response.value)


class LocalSource:
    def discover(self, client: SyncHttpClientProtocol) -> Sequence[Ref]:
        return ()


@dataclass(frozen=True)
class LocalCatalogPlugin:
    name: str = "local-catalog"
    source: LocalSource = LocalSource()
    expanders: tuple[CategoryList, ItemList] = (CategoryList(), ItemList())
    sink: ItemSink = ItemSink()


def crawl_one(
    top_ref: Ref, plugin: LocalCatalogPlugin, client: SyncHttpClientProtocol
) -> RunResult | None:
    try:
        result = run_crawl(
            top_ref, cast("CrawlPlugin", plugin), client, RunConfig()
        )
    except ExpansionNotReadyError as exc:
        print(f"retry next scheduled run: {exc}")
        return None
    except PartialExpansionError as exc:
        print(f"retry after the listing is complete: {exc}")
        return None

    print(f"persisted {result.leaves_persisted} item(s)")
    branch_errors = [
        error for error in result.errors if error.startswith("expander branch")
    ]
    if branch_errors:
        print(f"partial branches: {branch_errors}")
    return result


def run_example() -> RunResult:
    server, thread = build_mock_server()
    host, port = cast(tuple[str, int], server.server_address)
    base_url = f"http://{host}:{port}"
    plugin = LocalCatalogPlugin()
    try:
        with HttpClient(
            HttpClientConfig(user_agent="example-crawler/1.0")
        ) as client:
            crawl_one(Ref(f"{base_url}/top?mode=not-ready"), plugin, client)
            crawl_one(Ref(f"{base_url}/top?mode=partial"), plugin, client)
            result = crawl_one(Ref(f"{base_url}/top"), plugin, client)
        assert result is not None
        return result
    finally:
        server.shutdown()
        server.server_close()
        thread.join()


if __name__ == "__main__":
    run_example()

Leaf failures are different: the runner records LeafUnavailableError and other non-fatal exceptions from Sink.consume() in result.errors, increments leaves_failed, and continues with other leaves. Make persistence idempotent so a scheduled retry can safely revisit successful leaves too. Cancellation and explicitly fatal AssetDownloadError still propagate.

Async leaf processing for high throughput

Implement the async protocols (async def expand and async def consume) and pass an AsyncHttpClientProtocol implementation. Expanders are awaited in tree order; leaf consume() calls run concurrently up to async_concurrency.

import asyncio
import json
from collections.abc import Sequence
from dataclasses import dataclass
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread
from typing import cast

from ladon import (
    AsyncHttpClient,
    AsyncHttpClientProtocol,
    Expansion,
    HttpClientConfig,
    LeafUnavailableError,
    Ref,
    RunConfig,
    RunResult,
    async_run_crawl,
)


class AsyncCatalogHandler(BaseHTTPRequestHandler):
    def do_GET(self) -> None:  # noqa: N802
        host, port = cast(tuple[str, int], self.server.server_address)
        if self.path == "/listing":
            body = {
                "items": [
                    f"http://{host}:{port}/item/{number}" for number in range(3)
                ]
            }
        elif self.path.startswith("/item/"):
            body = {"id": int(self.path.rsplit("/", 1)[1])}
        else:
            self.send_error(404)
            return
        encoded = json.dumps(body).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(encoded)))
        self.end_headers()
        self.wfile.write(encoded)

    def log_message(self, format: str, *args: object) -> None:
        pass


def build_mock_server() -> tuple[ThreadingHTTPServer, Thread]:
    server = ThreadingHTTPServer(("127.0.0.1", 0), AsyncCatalogHandler)
    thread = Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, thread


class ListingExpander:
    async def expand(
        self, ref: Ref, client: AsyncHttpClientProtocol
    ) -> Expansion:
        response = await client.get(ref.url)
        if not response.ok or response.value is None:
            raise LeafUnavailableError(f"listing failed: {response.error}")
        payload = json.loads(response.value)
        return Expansion(
            record={"listing": ref.url},
            child_refs=[Ref(url) for url in payload["items"]],
        )


class ItemSink:
    async def consume(
        self, ref: Ref, client: AsyncHttpClientProtocol
    ) -> dict[str, object]:
        response = await client.get(ref.url)
        if not response.ok or response.value is None:
            raise LeafUnavailableError(f"item failed: {response.error}")
        return json.loads(response.value)


class LocalSource:
    async def discover(self, client: AsyncHttpClientProtocol) -> Sequence[Ref]:
        return ()


@dataclass(frozen=True)
class AsyncCatalogPlugin:
    name: str = "async-local-catalog"
    source: LocalSource = LocalSource()
    expanders: tuple[ListingExpander, ...] = (ListingExpander(),)
    sink: ItemSink = ItemSink()


async def crawl_one(base_url: str) -> RunResult:
    persisted: list[dict[str, object]] = []

    async def persist(
        leaf_record: dict[str, object], parent_record: object
    ) -> None:
        del parent_record
        persisted.append(leaf_record)

    config = HttpClientConfig(user_agent="my-async-crawler/1.0", retries=0)
    async with AsyncHttpClient(config) as client:
        result = await async_run_crawl(
            top_ref=Ref(f"{base_url}/listing"),
            plugin=AsyncCatalogPlugin(),
            client=client,
            config=RunConfig(leaf_limit=500, async_concurrency=20),
            on_leaf=persist,
        )
    print(result.leaves_consumed, result.leaves_failed, persisted)
    return result


def run_example() -> RunResult:
    server, thread = build_mock_server()
    host, port = cast(tuple[str, int], server.server_address)
    try:
        return asyncio.run(crawl_one(f"http://{host}:{port}"))
    finally:
        server.shutdown()
        server.server_close()
        thread.join()


if __name__ == "__main__":
    run_example()

on_leaf must be an async def callback. Keep its work bounded as well: each concurrency slot spans both sink.consume() and the callback.

Respect robots.txt on a public-web crawl

For third-party public sites, enable robots enforcement on the sync client before discovery. Ladon fetches and caches each origin's robots.txt, blocks disallowed URLs as RobotsBlockedError, and honours Crawl-delay.

client.get() returns a blocked request in response.error; plugins must check for RobotsBlockedError and re-raise it so run_crawl() callers can handle it. See PublicPage.expand() in examples/cookbook/robots_txt.py.

import json
from collections.abc import Sequence
from dataclasses import dataclass
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread
from typing import cast

from ladon import (
    ChildListUnavailableError,
    CrawlPlugin,
    Expansion,
    HttpClient,
    HttpClientConfig,
    LeafUnavailableError,
    Ref,
    RobotsBlockedError,
    RunConfig,
    RunResult,
    SyncHttpClientProtocol,
    run_crawl,
)


class RobotsHandler(BaseHTTPRequestHandler):
    def do_GET(self) -> None:  # noqa: N802
        if self.path == "/robots.txt":
            body = b"User-agent: example-crawler\nDisallow: /blocked\n"
        elif self.path == "/allowed":
            body = json.dumps({"item": "/item"}).encode()
        elif self.path == "/item":
            body = json.dumps({"title": "allowed local result"}).encode()
        elif self.path == "/blocked":
            body = b"this response must never be fetched"
        else:
            self.send_error(404)
            return
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format: str, *args: object) -> None:
        pass


def build_mock_server() -> tuple[ThreadingHTTPServer, Thread]:
    server = ThreadingHTTPServer(("127.0.0.1", 0), RobotsHandler)
    thread = Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, thread


class PublicPageSource:
    def __init__(self, base_url: str) -> None:
        self._base_url = base_url

    def discover(self, client: SyncHttpClientProtocol) -> Sequence[Ref]:
        return [
            Ref(f"{self._base_url}/allowed"),
            Ref(f"{self._base_url}/blocked"),
        ]


class PublicPage:
    def expand(self, ref: Ref, client: SyncHttpClientProtocol) -> Expansion:
        response = client.get(ref.url)
        if isinstance(response.error, RobotsBlockedError):
            raise response.error
        if not response.ok or response.value is None:
            raise ChildListUnavailableError(f"page failed: {response.error}")
        payload = json.loads(response.value)
        return Expansion(
            record={"page": ref.url},
            child_refs=[Ref(f"{ref.url.rsplit('/', 1)[0]}{payload['item']}")],
        )


class PublicPageSink:
    def consume(
        self, ref: Ref, client: SyncHttpClientProtocol
    ) -> dict[str, object]:
        response = client.get(ref.url)
        if not response.ok or response.value is None:
            raise LeafUnavailableError(f"item failed: {response.error}")
        return json.loads(response.value)


@dataclass(frozen=True)
class PublicPagePlugin:
    name: str
    source: PublicPageSource
    expanders: tuple[PublicPage, ...] = (PublicPage(),)
    sink: PublicPageSink = PublicPageSink()


def run_example() -> list[RunResult]:
    server, thread = build_mock_server()
    host, port = cast(tuple[str, int], server.server_address)
    base_url = f"http://{host}:{port}"
    plugin = PublicPagePlugin("robots-local-site", PublicPageSource(base_url))
    results: list[RunResult] = []
    config = HttpClientConfig(
        user_agent="example-crawler/1.0",
        respect_robots_txt=True,
        min_request_interval_seconds=0.0,
    )
    try:
        with HttpClient(config) as client:
            for top_ref in plugin.source.discover(client):
                try:
                    result = run_crawl(
                        top_ref,
                        cast("CrawlPlugin", plugin),
                        client,
                        RunConfig(leaf_limit=100),
                    )
                except RobotsBlockedError as exc:
                    print(f"skipped {top_ref.url}: {exc}")
                    continue
                results.append(result)
                print(f"saved {result.leaves_consumed} result(s)")
        return results
    finally:
        server.shutdown()
        server.server_close()
        thread.join()


if __name__ == "__main__":
    run_example()

AsyncHttpClient currently raises NotImplementedError if respect_robots_txt=True; use the sync client for crawls that require Ladon's built-in robots enforcement.