forktex-scraping

Acquire public data, and be able to prove what you acquired.

A library, not a framework. There is no base class to inherit, no spider to register, no runner that calls you back — every piece works on its own, and you use the ones you need.

pip install forktex-scraping[http]      # stateless: fetch, pace, retry, page
pip install forktex-scraping[browser]   # stateful: a live page, and what its JS asks for

Fetch one page, politely

Six lines, and the politeness is real: robots.txt is checked, the rate is held per-origin, and 429/5xx back off on a budget.

import asyncio
from forktex_scraping import HttpTransport, Politeness, PolitenessSpec, Request

async def main():
    async with HttpTransport() as http:
        response = await Politeness().request(
            http,
            Request(url="https://example.com"),
            PolitenessSpec(rate_per_second=1),
        )
        print(response.status, response.text()[:80])

asyncio.run(main())

PolitenessSpec() defaults to 1 request/second, robots respected, 3 attempts. Politeness is an object you construct and pass, never a global — so two harvests against two hosts in one process do not interfere, and it works inside an API request handler.

A non-2xx is not an exception. It comes back as a Response, because a 404 is an answer.

Watch what a page's JavaScript actually asks for

This is the part worth reaching for. Most sites you would scrape are a thin UI over an API they never document — and the page will tell you the whole thing if you watch it.

Hand it a callback and every call the page makes arrives as it happens. Map it into your own types, write it wherever you keep things, count it, drop it — the library holds nothing.

import asyncio
from forktex_scraping import BrowserSession

async def main():
    rows = []

    async def mine(call):                      # your types, your storage, your rules
        if "api/search" in call.request.url:
            rows.extend(call.response.json()["items"])

    async with BrowserSession(on_exchange=mine) as page:
        await page.goto("https://example.com/search")
        await asyncio.sleep(5)

    print(len(rows), "rows, mapped by you")

asyncio.run(main())

Scraping the rendered DOM gives you the twenty rows on screen. Reading the request that produced them gives you every row, forever, over plain HTTP — and no browser after the first look.

Nothing is buffered when you pass on_exchange. Without it the session keeps a list for you, which is convenient for a short exploration and is what exchanges() and wait_for_exchange() read — but it grows for the life of the session, and a consumer who maps each call and drops it should not pay for a list they never read.

async with BrowserSession() as page:           # buffering: fine for a look around
    await page.goto("https://example.com/search")

    call = await page.wait_for_exchange("api/search", timeout_ms=30_000)
    print(sorted(call.request.body))           # the filter names the UI actually sends

    for call in page.exchanges():
        ...

By default only the site's own API calls are recorded — xhr and fetch, not images and stylesheets. Say otherwise when you need to:

BrowserSession(record_types={"xhr", "fetch", "document"})   # chasing an attachment
BrowserSession(record_types=None)                           # everything

page.observe() returns a PageState — the URL, the title, the text with navigation chrome stripped, and (when buffering) the exchanges since your last look. Pass text_limit= if the default 50 000 characters is the wrong budget for you.

Page an API, with the cap defeated

harvest is for the case that motivates this library: a remote that answers a query it could not fully satisfy without saying so.

import asyncio, pathlib
from forktex_scraping import (
    ApiSource, AssertKind, FilterField, HttpTransport, JsonlSink,
    Politeness, PolitenessSpec, RequestContract, Tier, harvest,
)

source = ApiSource(
    id="example.search",
    tier=Tier.CAPTURED_CONTRACT,
    politeness=PolitenessSpec(rate_per_second=2),
)

contract = RequestContract(
    source="example.search",
    url="https://example.com/api/search",
    method="POST",
    headers={"content-type": "application/json"},
    body_template={"sort": []},
    # Where each filter's effect must show up in a returned row. This is the load-bearing part.
    filters=(FilterField(field="q", assert_path="title", assert_kind=AssertKind.CONTAINS_FOLDED),),
    page_field="page",
    size_field="pageSize",
    page_size=100,
    first_page=0,
    items_path="items",
    total_path="total",
    cap_signal="truncated",   # the flag this remote sets when it capped the answer
    cap_size=3000,
)

async def main():
    sink = JsonlSink(pathlib.Path("out.jsonl"))
    async with HttpTransport() as http:
        coverage = await harvest(
            source, contract,
            transport=http, politeness=Politeness(),
            sink=sink, run_id="run-1",
            filters={"q": "software"},
        )
    await sink.close()

    print(coverage.summary())
    if not coverage.complete:
        for window in coverage.truncated:      # named, never silently missing
            print("could not cover", window.label())

asyncio.run(main())

Three things happen that you did not have to write:

The cap is defeated. When the remote signals truncation, the window is split along its axis and re-fetched until each part comes back honest. A window that cannot be split further is reported in coverage.truncated — not dropped, and not quietly accepted.

Every filter is checked against every row. assert_path says where a filter's effect must appear, and assert_kind says how. A row that violates it raises ContractDriftError, because the remote answered a different question than the one you asked, and filtering client-side would hide that behind a plausible-looking result.

Pick the kind that describes what the remote does, not what you would prefer:

Kind	Use when the remote
CONTAINS	matches a substring, case-insensitively
CONTAINS_FOLDED	also ignores accents — searching platforma returns platformă
PREFIX, EQUALS	matches from the start, or exactly
GTE, LTE	bounds a value as an instant
DATE_GTE, DATE_LTE	bounds by calendar day rather than moment

The folded and date variants exist because a strict comparison twice called a correctly-honoured filter a violation. An assertion that is stricter than the remote aborts good harvests.

Coverage is a claim you can inspect. coverage.complete is derived from what is in truncated and failed, so it cannot disagree with them.

Why assert_path has no default

Because a remote can accept a filter field it does not recognise, ignore it, and answer 200 with its capped unfiltered set — byte-identical to sending no filter at all. Measured, on a public procurement portal: an invented field name returned exactly the same 3 000 rows as sending nothing. A guessed field name does not fail. It produces a confidently wrong dataset, which is worse than an error, and no amount of care in your own code detects it.

So a filter you cannot verify is one this library will not send.

The one distinction to hold

Not HTTP versus browser — whether a thing can be replayed.

	Transport (stateless)	Session (stateful)
snapshot and replay	yes — a request hashes to a stable identity	no — order is the state
retry, paginate, cap-split	yes	not applicable
run concurrently	yes, under the pacer	one session, one caller

The rule that follows: a session's output is never a result. It is data — a contract, a source definition — that the stateless path then replays forever. Explore statefully; harvest statelessly. That is what keeps a run reproducible.

Other shapes

Not every source is a paged API behind a captured contract:

await harvest_keys(source, keys, ..., payload_for=lambda batch: [{"id": k} for k in batch])
await harvest_catalog(source, ...)    # CKAN-shaped open data
await harvest_sitemap(source, ...)    # enumerate via declared sitemaps, robots honoured

harvest_keys takes a payload_for you write, because a batch body's shape is the one thing that is genuinely per-API — and that knowledge belongs in your code, not this library's.

What you plug in

Everything below is a Protocol. Nothing to inherit — any object with the right methods works, including one you already have.

Port	Methods	Ships with
Transport	fetch, close	HttpTransport
Sink	emit, close	JsonlSink, RunDirectory.sink
SnapshotStore	put, get	RunDirectory
Session	goto, act, observe, exchanges, close	BrowserSession

Sink.emit takes a plain dict, deliberately: what a record means is your domain, and this library must not learn it.

Want runs on disk with verbatim evidence? RunDirectory(root) is both a sink and a snapshot store, writing <run-id>/records.jsonl, <run-id>/raw/<hash>.json.gz and a manifest. Want something else? Pass your own object. Want no snapshots? Leave the argument out.

Politeness, stated plainly

robots.txt is honoured and fails open — an unreachable or malformed one is not a refusal, because treating absence as prohibition makes the polite configuration the one that cannot fetch anything. A genuine Disallow raises RobotsRefusedError rather than logging: a politeness module that only warns is one that ships the violation.

The pacer is per-origin and shared across callers, so eight concurrent workers at "one per second" produce one per second, not eight. Retry-After wins over the computed backoff.

There is no stealth. No user-agent rotation, no proxy pool, no fingerprint patching. The agent string says who you are. A package that shipped evasion machinery would be used for evasion.

Errors

Every one is a leaf of forktex.error.AppError, so one except covers them, and each names a condition rather than a status code:

TransportError · ThrottledError · RobotsRefusedError · ContractDriftError · PartitionExhaustedError · ExtractionError · SourceDefinitionError · UnknownSourceError

ContractDriftError is the load-bearing one. It means the contract no longer describes the remote, so it aborts the run rather than being noted per-partition — every other partition is equally suspect.

Requirements

Python 3.14+. Extras import lazily: import forktex_scraping works with none of them installed, and asking for one you did not install raises an ImportError that names the extra.

Extra	Pulls	For
[http]	httpx	the stateless transport — almost everything
[browser]	playwright	watching a page, once per source
[html]	selectolax	parsing a server-rendered table

Documentation

One page per module in docs/, generated from typed JSON records under docs/sources/ — edit the JSON, not the markdown. docs/design.md is why it is shaped this way.

Licence

Dual-licensed: AGPL-3.0-or-later, or a commercial licence for use in proprietary products and SaaS deployments where AGPL obligations cannot be met — info@forktex.com. See LICENSE and NOTICE.