> ## Documentation Index
> Fetch the complete documentation index at: https://docs.context.dev/llms.txt
> Use this file to discover all available pages before exploring further.
# Build a refreshable dataset from websites
> Discover source pages, extract nullable records, and update a durable dataset without duplicating rows or erasing good data after failures.
export const AgentSetupPrompt = ({children, variant = "setup"}) => {
const isRecipe = variant === "recipe";
const promptLabel = isRecipe ? "recipe prompt" : "setup prompt";
const [copyState, setCopyState] = useState("idle");
const contentRef = useRef(null);
const detailsRef = useRef(null);
const timerRef = useRef(null);
const mountedRef = useRef(true);
const pendingRef = useRef(false);
useEffect(() => {
mountedRef.current = true;
return () => {
mountedRef.current = false;
if (timerRef.current !== null) clearTimeout(timerRef.current);
};
}, []);
const copyPrompt = useCallback(async () => {
if (pendingRef.current) return;
pendingRef.current = true;
if (timerRef.current !== null) {
clearTimeout(timerRef.current);
timerRef.current = null;
}
setCopyState("copying");
try {
const code = contentRef.current?.querySelector("pre code") ?? contentRef.current?.querySelector("pre");
const prompt = code?.textContent;
if (typeof prompt !== "string" || !prompt.trim()) {
throw new Error("Prompt is not available.");
}
if (typeof navigator === "undefined" || !navigator.clipboard?.writeText) {
throw new Error("Clipboard access is not available.");
}
await navigator.clipboard.writeText(prompt);
if (!mountedRef.current) return;
setCopyState("copied");
timerRef.current = setTimeout(() => {
if (mountedRef.current) setCopyState("idle");
timerRef.current = null;
}, 4000);
} catch {
if (!mountedRef.current) return;
if (detailsRef.current) detailsRef.current.open = true;
setCopyState("error");
} finally {
pendingRef.current = false;
}
}, []);
const feedback = copyState === "copied" ? `${isRecipe ? "Recipe" : "Setup"} prompt copied. Paste it into your coding agent.` : copyState === "error" ? "Couldn't copy automatically. The prompt is open below. Select and copy the text manually." : "";
return
{isRecipe ? "Have your agent implement this recipe" : "Set up Context.dev"}
{isRecipe ? "Copy this prompt into your coding agent to build the recipe in your project." : "Paste into your coding agent. It will ask about your goal and project before making changes."}
{feedback}
{`View ${promptLabel}`}
{children}
;
};
```text Recipe prompt theme={null}
Implement this recipe in my project:
https://docs.context.dev/use-cases/structured-web-datasets
Read the recipe and linked API guides, inspect this project's stack, and build the complete flow using its existing conventions.
Build a resumable pipeline from reviewed source URLs through discovery, structured extraction, validation, and durable storage. Use stable record IDs, nullable schemas, fact checking, and explicit URL and variant identity. Store attempts separately from accepted records and observations, bound retries, and retain the last good record when extraction or validation fails. Support refreshes without duplicating records.
Reuse existing Context.dev configuration and keep secret API keys on the server. Add focused tests, run the relevant checks, and document setup and how to try the result.
```
Turn a reviewed set of web pages into a dataset that you can resume, audit, and refresh. Context.dev collects or extracts the source data; your application owns stable record IDs, validation, normalization, and storage.
[Murph built a food and supplement database with Context.dev](https://www.context.dev/blog/murph-builds-a-2m-food-and-supplement-database-with-context-dev), while [Bystreet uses it for large-scale scraping](https://www.context.dev/blog/bystreet-completed-1m-scrapes-with-context-dev-without-any-issues). This recipe starts with a small product catalog stored in SQLite.
## Discover and checkpoint the source list
Use [Sitemap](/guides/discover-website-urls) to find candidate URLs, or a [scoped crawl](/guides/crawl-website) when you need linked content. Review the domain and path scope before queuing pages. Give each item an application ID that stays stable if its title or price changes.
Save a manifest like this as `sources.json`, replacing the example URL with a reviewed product page:
```json sources.json theme={null}
[
{ "id": "supplier-a:mug-blue", "url": "https://shop.example.com/products/blue-mug" }
]
```
Use one item per intended record or variant. Do not assume different query parameters identify the same product: a query can select a size, color, locale, or currency. Keep an explicit canonical-URL policy and retain the original source URL.
## Choose the extraction path
| Dataset requirement | API |
| ----------------------------------------- | --------------------------------------------------------------------------------- |
| Standard fields from a known product page | [Product](/guides/extract-product-from-websites) |
| Your own fields or non-product records | [Extract](/guides/extract-structured-data-from-websites) with a JSON Schema |
| Large collection of raw Markdown or HTML | [Batch](/guides/scrape-websites-in-batches), followed by your processing pipeline |
The Products discovery endpoint returns at most 12 products; it is not a full-catalog export. For a catalog, discover and queue the pages you intend to cover. Product variants, stock, and image-to-variant associations can be missing and need independent validation.
The worker below uses custom Extract with `factCheck: true`, a nullable schema, and a single-page scope. See the [extraction guide](/guides/extract-structured-data-from-websites#make-an-extraction-request) for requests in every SDK. Run this Python application with `CONTEXT_DEV_API_KEY` set on the server.
## Store records separately from attempts
Keep three kinds of state: the source queue, the latest accepted record, and historical observations. The SQLite worker commits each source independently, so a failure on one page cannot delete another page's data.
```python catalog.py theme={null}
import json
import math
import os
import sqlite3
import sys
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import Request, urlopen
SCHEMA = {
"type": "object",
"properties": {
"name": {"type": ["string", "null"], "description": "Stated product name; null if absent."},
"variant": {"type": ["string", "null"], "description": "Selected size, color, or variant; null if unclear."},
"price_amount": {"type": ["number", "null"], "description": "Current stated price for this variant; null if unclear."},
"currency": {"type": ["string", "null"], "description": "Explicit three-letter currency code; null if unclear."},
"unit": {"type": ["string", "null"], "description": "Price basis, such as per item or per kilogram; null if absent."},
},
"required": ["name", "variant", "price_amount", "currency", "unit"],
"additionalProperties": False,
}
def validate_record(data):
# Local validation for the small schema above; reject booleans as prices.
if not isinstance(data, dict) or set(data) != set(SCHEMA["required"]):
raise ValueError("Unexpected record shape")
for key in ("name", "variant", "currency", "unit"):
if data[key] is not None and not isinstance(data[key], str):
raise ValueError(f"Invalid {key}")
price = data["price_amount"]
if price is not None and (type(price) not in (int, float) or not math.isfinite(price) or price < 0):
raise ValueError("Invalid price")
if not data["name"] or not data["name"].strip():
raise ValueError("No usable product identity")
normalized = dict(data)
if normalized["currency"] is not None:
normalized["currency"] = normalized["currency"].strip().upper()
if len(normalized["currency"]) != 3 or not normalized["currency"].isascii() or not normalized["currency"].isalpha():
raise ValueError("Currency needs review")
return normalized
def open_catalog(path):
db = sqlite3.connect(path)
db.row_factory = sqlite3.Row
db.executescript("""
CREATE TABLE IF NOT EXISTS sources (
id TEXT PRIMARY KEY, url TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending',
attempts INTEGER NOT NULL DEFAULT 0, next_attempt_at REAL NOT NULL DEFAULT 0,
error TEXT, last_success_at TEXT
);
CREATE TABLE IF NOT EXISTS records (
id TEXT PRIMARY KEY, source_url TEXT NOT NULL, observed_at TEXT NOT NULL,
data_json TEXT NOT NULL, analyzed_urls_json TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS observations (
id TEXT NOT NULL, observed_at TEXT NOT NULL, source_url TEXT NOT NULL,
raw_data_json TEXT NOT NULL, analyzed_urls_json TEXT NOT NULL,
PRIMARY KEY (id, observed_at)
);
""")
return db
def enqueue(db, items, refresh=False):
with db:
for item in items:
url = urlsplit(item["url"])
if url.scheme != "https" or not url.hostname or url.username or url.password:
raise ValueError("Use reviewed HTTPS source URLs")
db.execute("""
INSERT INTO sources (id, url) VALUES (?, ?)
ON CONFLICT(id) DO UPDATE SET
status=CASE WHEN url <> excluded.url THEN 'pending' ELSE status END,
attempts=CASE WHEN url <> excluded.url THEN 0 ELSE attempts END,
next_attempt_at=CASE WHEN url <> excluded.url THEN 0 ELSE next_attempt_at END,
url=excluded.url
""", (item["id"], item["url"]))
if refresh:
db.execute("UPDATE sources SET status='pending', attempts=0, next_attempt_at=0, error=NULL WHERE id=?", (item["id"],))
def extract_page(url):
request = Request(
"https://api.context.dev/v1/web/extract",
data=json.dumps({
"url": url, "schema": SCHEMA, "factCheck": True,
"maxPages": 1, "maxDepth": 0, "maxAgeMs": 0,
}).encode(),
headers={"Authorization": f"Bearer {os.environ['CONTEXT_DEV_API_KEY']}",
"Content-Type": "application/json"},
method="POST",
)
with urlopen(request, timeout=120) as response:
return json.load(response)
def retry_delay(headers):
value = headers.get("Retry-After")
try:
return max(0, float(value))
except (TypeError, ValueError):
try:
return max(0, parsedate_to_datetime(value).timestamp() - time.time())
except (TypeError, ValueError, AttributeError):
return 30
def run_pending(db):
jobs = db.execute("""
SELECT * FROM sources WHERE status IN ('pending', 'retryable')
AND attempts < 3 AND next_attempt_at <= ?
""", (time.time(),)).fetchall()
for job in jobs:
with db:
db.execute("UPDATE sources SET attempts=attempts+1 WHERE id=?", (job["id"],))
try:
result = extract_page(job["url"])
data = validate_record(result["data"])
urls = result["urls_analyzed"]
if not isinstance(urls, list) or not urls or not all(isinstance(url, str) for url in urls):
raise ValueError("No source provenance")
observed = datetime.now(timezone.utc).isoformat()
with db:
db.execute("INSERT INTO observations VALUES (?, ?, ?, ?, ?)",
(job["id"], observed, job["url"], json.dumps(result["data"]), json.dumps(urls)))
db.execute("""
INSERT INTO records VALUES (?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET source_url=excluded.source_url,
observed_at=excluded.observed_at, data_json=excluded.data_json,
analyzed_urls_json=excluded.analyzed_urls_json
""", (job["id"], job["url"], observed, json.dumps(data), json.dumps(urls)))
db.execute("UPDATE sources SET status='ready', error=NULL, last_success_at=? WHERE id=?", (observed, job["id"]))
except HTTPError as error:
retryable = error.code in (408, 429) or error.code >= 500
delay = retry_delay(error.headers)
error.close()
with db:
db.execute("UPDATE sources SET status=?, error=?, next_attempt_at=? WHERE id=?",
("retryable" if retryable else "error", f"HTTP {error.code}",
time.time() + delay, job["id"]))
if error.code in (401, 403, 429):
break
except (URLError, TimeoutError):
with db:
db.execute("UPDATE sources SET status='retryable', error='Network failure', next_attempt_at=? WHERE id=?", (time.time() + 30, job["id"]))
except (ValueError, KeyError, TypeError) as error:
with db:
db.execute("UPDATE sources SET status='review', error=? WHERE id=?", (str(error), job["id"]))
if __name__ == "__main__":
with open(sys.argv[1]) as source_file:
manifest = json.load(source_file)
db = open_catalog("catalog.sqlite")
enqueue(db, manifest, refresh="--refresh" in sys.argv[2:])
run_pending(db)
for row in db.execute("SELECT status, count(*) AS count FROM sources GROUP BY status"):
print(dict(row))
db.close()
```
```bash Run the worker theme={null}
python3 catalog.py sources.json
python3 catalog.py sources.json --refresh
```
The first command resumes pending or eligible retryable items and skips completed ones. The second starts a new observation cycle for the listed sources. Because the application IDs are stable, a changed price updates the current record instead of creating a duplicate product. Each accepted refresh also records an observation for history.
This is a single-worker example. Use queue leases or transactional job claiming before running multiple workers. Schedule retries after `next_attempt_at`; do not repeatedly restart after an authentication or account-limit error. Review exhausted retries and invalid records explicitly.
## Preserve uncertainty and provenance
`urls_analyzed` is the set of pages used by an extraction request, not automatic field-level evidence. If a displayed field needs an exact citation, retain and verify a supporting excerpt from the relevant page. Do not manufacture a per-field source by attaching the first analyzed URL to every value.
A null price means the extraction could not establish it. Keep historical observations if you want to display a separately labeled last-known price. Never replace an unknown price with zero, or compare prices before checking currency, unit, and selected variant.
The worker uses `maxAgeMs: 0` for an intentional fresh extraction. For routine ingestion, choose a cache policy that fits your refresh schedule. The stored observation time is when your worker received the data; inspect cache metadata if source freshness is material.
## Expand and reconcile coverage
For larger raw-content jobs, use Batch and persist the batch ID, per-item IDs, result cursor, and failures. Read all result pages using `has_more` and `next_cursor`; a completed job can still contain failed items. Resume from checkpoints and process each result idempotently.
Before calling a dataset complete, compare discovered, queued, accepted, empty, failed, and reviewed items. A successful extraction or a crawl that reached its time or page limit does not establish full-site coverage. Remove a record only after a deliberate deletion check; a failed page is not evidence that the product disappeared.
Try a second run with a changed price, a nullable field, and one failing source. Verify that IDs remain unique, observations retain source URLs and times, and the failed source's last accepted record remains available with a stale-state label.
Use the product-specific contract when it fits your records.
Compare compatible facts and preserve unknown values.