Python SDK

install megalania, then Megalania().scrape(url) (the key comes from MEG_API_KEY). You get the body, the response headers as typed fields, and a MegalaniaError with the error code and message. It retries once on a 429, after Retry-After, and never on anything else. Every scrape, map and create call carries an Idempotency-Key, so a repeat is never charged twice.

The SDK is in beta. Its types are generated from the OpenAPI document (api/openapi/v1.yaml), so they cannot drift from the API. The code is MIT licensed at https://github.com/megalania/megalania-python. The package also installs the megalania command, and it is the base of the LangChain and LlamaIndex integrations.

Install

Python 3.10 or newer.

pip install megalania

Scrape, extract and batch the Lab

Three first calls against the Lab, our test site: one page, rows from a list page, and a batch of both. Each one runs as pasted once MEG_API_KEY is set. The dashboard’s SDKs & code page shows the same code.

One page as Markdown

import os

from megalania import Megalania

with Megalania(os.environ["MEG_API_KEY"], base_url="https://get.megalania.io", api_url="https://api.megalania.io") as meg:
    page = meg.scrape("https://lab.megalania.io/article", response_types="markdown")
    print(page.headers.credits_charged, page.text[:300])

Rows from a list page

import os

from megalania import Megalania

with Megalania(os.environ["MEG_API_KEY"], base_url="https://get.megalania.io", api_url="https://api.megalania.io") as meg:
    rows = meg.scrape("https://lab.megalania.io/cards", response_types="json", extract="items:.card{title:h2|price:.price|url:a@href}*")
    print(rows.headers.credits_charged, rows.text[:300])

Many pages in one batch

import os

from megalania import Megalania

with Megalania(os.environ["MEG_API_KEY"], base_url="https://get.megalania.io", api_url="https://api.megalania.io") as meg:
    batch = meg.create_batch(["https://lab.megalania.io/article", "https://lab.megalania.io/cards"], response_types="markdown")
    print(batch["id"], meg.get_batch(batch["id"])["status"])

One page

import os

from megalania import Megalania, MegalaniaError

base_url = os.environ.get("MEG_BASE_URL", "https://get.megalania.io")
with Megalania(os.environ["MEG_API_KEY"], base_url=base_url) as meg:
    try:
        page = meg.scrape("https://example.com/", country_code="US")
    except MegalaniaError as e:
        print(e.status, e.code, e.field, e.message, e.request_id)
        raise
    print(page.headers.credits_charged, page.headers.target_status, page.headers.request_id)
    print(page.text[:200])

scrape(url, **params) takes the keys of the JSON envelope: engine, country_code, proxy_tier, response_types, extract, max_credits and the rest, as keyword arguments. It sends only the keys you pass. A key you leave out gets the server’s default, which is not always the same as sending that default yourself: a proxy_tier you leave out lets us pick the cheapest route that works.

scrape sends an Idempotency-Key: a fresh UUID, or the idempotency_key="..." you pass. Pass your own on a call you might send again yourself. A repeat is a 409 replayed, never a second charge.

Batches, monitors, map and tier

Batches and monitors live on the API host, https://api.megalania.io. Pass api_url to change it. Each method is named after its REST operation and returns the JSON answer as a dict.

import os

from megalania import Megalania

with Megalania(os.environ["MEG_API_KEY"]) as meg:
    print(meg.get_tier("docs.megalania.io"))
    print(meg.map_site("https://docs.megalania.io/start/quickstart", limit=20))
    batch = meg.create_batch(["https://docs.megalania.io/start/quickstart"], response_types="markdown")
    print(meg.get_batch(batch["id"]))
    print(meg.list_batch_items(batch["id"]))
    monitor = meg.create_monitor(name="Demo", urls=["https://docs.megalania.io/start/quickstart"], every="1d")
    meg.update_monitor(monitor["id"], paused=True)

scrape, map_site, create_batch and create_monitor send an Idempotency-Key, yours or a fresh one. The 429 rule and MegalaniaError are the same as for scrape.

Read many pages

wait_batch(id) polls until a batch is done, failed or canceled. batch_results(id) then yields each finished item with its body. It fetches several result links at once, and it never sends your key to them. read_pages(urls) scrapes one page at a time, or crawls from one URL with crawl={...}. It yields (text, metadata). With a schema, it asks for json, and the text is the filled schema.

import os

from megalania import Megalania

with Megalania(os.environ["MEG_API_KEY"]) as meg:
    batch = meg.create_batch(["https://docs.megalania.io/start/quickstart"], response_types="markdown")
    print(meg.wait_batch(batch["id"])["status"])
    for item, body in meg.batch_results(batch["id"]):
        print(item["url"], len(body))
    schema = {"type": "object", "additionalProperties": False, "properties": {"title": {"type": "string"}}}
    for text, meta in meg.read_pages("https://docs.megalania.io/start/quickstart", schema=schema):
        print(meta["credits_charged"], text[:200])

Account, usage and stats

get_account() shows your balance and limits, with either key. get_domain_stats(host) shows how the whole platform does on one host. It is free. get_usage(days=7) and rotate_webhook_secret() need an account key (meg_ak_). The new secret shows once, so store it right away.

import os

from megalania import Megalania

with Megalania(os.environ["MEG_API_KEY"]) as meg:
    print(meg.get_account())
    print(meg.get_domain_stats("docs.megalania.io"))
with Megalania(os.environ["MEG_ACCOUNT_KEY"]) as meg:
    print(meg.get_usage(days=7))
    secret = meg.rotate_webhook_secret()

Countries

GET /v1/geos needs no key. The generated client has it as get_geos:

from megalania._generated import Client
from megalania._generated.api.default import get_geos

geos = get_geos.sync(client=Client(base_url="https://get.megalania.io"))
for route in geos.routes:
    print(route.route, route.proxy_tier, len(route.countries))

A country_code the route does not serve is a free 400 with field country_code. Countries has the full list.

What you get back

scrape returns a Result:

The SDK reads the whole body into memory. A page can be up to 50 MB. A call waits at most timeout seconds for each step, the connection and each read (100 by default, Megalania(key, timeout=...)), then raises httpx.TimeoutException.

Errors

Any answer that is not a 2xx raises MegalaniaError. A page the site blocked is a free 422 with code blocked: you were not charged, and the error is the whole answer, with no page content. It has status, code (the error.code, such as bad_parameter), message, field (the parameter to change, on 400 and 422), request_id, report_url and headers (the same Headers a result has, so a block shows untried, hops and vendor). Error codes lists every code. If the body is not our JSON, for example a page from a proxy of yours, code is None and message holds the start of that body.

Retries

A 429 over_concurrency is retried once, after the seconds in Retry-After (1 when the header is missing), with the same Idempotency-Key. A 429 releases the key, so the retry is safe. A second 429 raises, and so does a 429 whose Retry-After is above 60 seconds or not a whole number of seconds, without waiting. Nothing else is retried: a 503 can mean the fetch ran, so you decide whether to send it again.