scrapedatshi-py

Official Python SDK for the scrapedatshi RAG pipeline API.

Scrape URLs, chunk documents, embed content, inject into vector databases, and extract structured data — all from a clean, typed Python interface.

Installation

pip install scrapedatshi

Requires Python 3.10+.

Quick Start

from scrapedatshi import ScrapedatshiClient

client = ScrapedatshiClient(api_key="sds_...")

result = client.pipeline.chunk_url("https://docs.example.com")
print(f"Got {result.total_chunks} chunks — cost ${result.credits_used:.4f}")
for chunk in result.chunks:
    print(chunk.content[:80])

CLI — Project Scaffolding

The SDK ships with a scrapedatshi CLI command that generates a ready-to-run sandbox project with pre-configured example scripts for every pipeline method.

scrapedatshi init my-project

This creates:

my-project/
├── .env                        ← add your API keys here (gitignored)
├── .gitignore
├── README.md
└── examples/
    ├── 00_discover_providers.py   ← list all providers + required fields (no keys needed)
    ├── 01_scrape_url.py
    ├── 02_pdf_extract.py
    ├── 03_scrape_file.py
    ├── 04_chunk_url.py
    ├── 05_chunk_file.py
    ├── 06_crawl_site.py
    ├── 07_sync_to_vdb.py
    ├── 08_ingest_file.py
    ├── 09_ingest_scraped.py
    ├── 10_autorag.py
    ├── 11_schema_extract.py
    ├── 12_extract_crawl.py
    ├── 13_query_vdb.py
    ├── 14_rag_chat.py
    ├── 15_inspect_vdb.py
    └── 16_capture_session.py

Each script has a clearly marked # ── CONFIGURE ── block at the top — just fill in your target URL, file path, or keys and run it. Start with 00_discover_providers.py to see all supported providers and the env vars each one needs.

cd my-project
python examples/00_discover_providers.py
python examples/01_scrape_url.py

Output files: Content-returning scripts (01, 02, 03, 08–11) automatically save results as JSON next to the script. Credits and job stats always print to the terminal. The filename auto-increments if it already exists (chunks.json → chunks(1).json) so no run overwrites a previous result. Set SAVE_TO = None in any script to print everything to the terminal instead.

Authentication

Pass your API key directly or set the SCRAPEDATSHI_API_KEY environment variable:

export SCRAPEDATSHI_API_KEY="sds_..."

# Explicit key
client = ScrapedatshiClient(api_key="sds_...")

# From environment variable
client = ScrapedatshiClient()

Get your API key at scrapedatshi.com/portal/register.

Fetch Mode

The SDK supports two fetch modes, controlled by the fetch_mode parameter on ScrapedatshiClient.

fetch_mode="local" (default)

The SDK fetches the URL on your machine using your IP address, then submits the raw HTML to our server for processing.

• ✅ Your IP is used — not our server's
• ✅ Billed at the standard per-URL rate ($0.0020)
• ✅ Faster — no double-hop latency

client = ScrapedatshiClient(api_key="sds_...")  # local fetch by default

fetch_mode="server"

Our server fetches the URL. Use this if you are behind a corporate firewall or need server-managed IP rotation.

• ⚠️ Our server's IP is used
• ⚠️ Billed at 2× the standard rate ($0.0040 / URL)
• ✅ Works from restricted environments

client = ScrapedatshiClient(api_key="sds_...", fetch_mode="server")

Scrape to Markdown

The simplest way to get clean text from any URL or local file — no chunking, no embedding, no vector DB required. Returns the full page or file content as clean Markdown.

# Optional: uncomment to target a specific section
# SELECTOR = "article"   # detected sections are printed below — copy one here

# Scrape a URL → Markdown
result = client.pipeline.scrape_url(
    "https://docs.example.com",
    # selector=SELECTOR,      # uncomment after choosing a section below
    # js_render=True,         # headless Chromium for SPAs ($0.0050/URL surcharge)
    # cookies={"session": "abc123"},   # authenticated scraping
    # headers={"Authorization": "Bearer eyJ..."},
)
print(result.markdown)
print(f"Cost: ${result.credits_used:.4f}")
# Detected content sections — uncomment SELECTOR above and re-run to target one:
if result.selectors_found:
    print(f"Sections: {result.selectors_found}")

# Scrape a local file → Markdown
result = client.pipeline.scrape_file("./docs/manual.pdf")
print(result.markdown)
print(f"Cost: ${result.credits_used:.4f}")

ScrapeResult model:

result.markdown          # str — full page/file content as clean Markdown
result.source            # str — URL or filename
result.title             # str | None — page title (None for local files)
result.selectors_found   # list[str] — CSS selectors for detected content sections
result.content_truncated # bool — True if content exceeded ~75,000 words
result.credits_used      # float
result.credits_remaining # float

selectors_found is always empty for scrape_file() — CSS selectors are an HTML concept.

Scrape = Markdown. Chunk = Chunks. Use scrape_url() / scrape_file() when you want the raw text. Use chunk_url() / chunk_file() when you want RAG-optimized segments ready for embedding.

PDF Extract

Extract clean text or structured tables from any PDF — by URL or local file. No chunking, no embedding, no vector DB required.

Billing:

• File upload: $0.0020 per request
• URL fetch: $0.0040 per request (server fetches the PDF)

Extract text from a PDF URL

result = client.pipeline.pdf_extract(url="https://example.com/annual-report.pdf")
print(result.text)
print(f"Cost: ${result.credits_used:.4f}")

Extract text from a local PDF file

result = client.pipeline.pdf_extract(file_path="./docs/manual.pdf")
print(result.text)
print(f"Cost: ${result.credits_used:.4f}")

Extract tables from a PDF

result = client.pipeline.pdf_extract(
    url="https://example.com/data.pdf",
    mode="tables",
)
for table in result.tables or []:
    print(table)

PdfExtractResult model:

result.source            # str — URL or filename
result.mode              # str — "text" or "tables"
result.text              # str | None — Markdown text (mode="text")
result.tables            # list | None — structured table data (mode="tables")
result.credits_used      # float
result.credits_remaining # float

Parameters:

Exactly one of url or file_path must be provided.

Async version:

result = await client.pipeline.pdf_extract_async(url="https://example.com/report.pdf")

Chunk to JSON

No embedding or vector DB required. Returns structured JSON chunks from any source.

Chunk a URL

result = client.pipeline.chunk_url("https://docs.example.com")

print(f"Got {result.total_chunks} chunks — cost ${result.credits_used:.4f}")
for chunk in result.chunks:
    print(chunk.content[:80])

Optional parameters:

result = client.pipeline.chunk_url(
    "https://docs.example.com/guide",
    selector="article",      # CSS selector to target main content
    chunk_size=512,           # tokens per chunk (default: 512)
    overlap=50,               # token overlap between chunks (default: 50)
    js_render=True,           # headless Chromium for SPAs
)

Hierarchical (Parent-Child) Chunking

Hierarchical chunking produces small child chunks (~128 tokens) for precise vector matching, each carrying a larger parent chunk (~512 tokens) as metadata. When a child chunk matches a query, the LLM receives the full parent chunk for context — dramatically improving cross-referencing accuracy.

result = client.pipeline.chunk_url(
    "https://docs.example.com",
    hierarchical=True,        # enable parent-child chunking
    child_chunk_size=128,     # child chunk size (default: 128 tokens)
    chunk_size=512,           # parent chunk size (default: 512 tokens)
)

print(f"Got {result.total_chunks} child chunks (hierarchical={result.hierarchical})")
for chunk in result.chunks:
    print(f"Child ({chunk.token_estimate} tokens): {chunk.content[:80]}")
    if chunk.parent_text:
        print(f"Parent ({chunk.parent_index}): {chunk.parent_text[:120]}")

When to use hierarchical chunking:

• Long-form documentation where a single sentence needs surrounding context to be meaningful
• Multi-hop questions that require cross-referencing information across sections
• Any corpus where small chunks improve search precision but large chunks improve answer quality

How to use the results in your RAG pipeline:

• Embed chunk.content (the small child chunk) into your vector DB
• Feed chunk.parent_text (the large parent chunk) to the LLM as context on retrieval
• The parent_text is also stored in vector DB metadata automatically when using sync() or ingest()

Chunk a PDF URL

Pass any PDF URL directly — S3 links, CDN URLs, direct .pdf links — and the API automatically detects and extracts text. No special parameters needed.

result = client.pipeline.chunk_url(
    "https://my-bucket.s3.amazonaws.com/reports/annual-report-2024.pdf"
)
print(f"Got {result.total_chunks} chunks from PDF")

Chunk a local file

Supports PDF, MD, TXT, YAML, YML, JSON, CSV, XLSX, DOCX, IPYNB, HTML, XML, and all common code files (.py, .js, .ts, .sql, .go, .rb, .java, etc.). In local-fetch mode (default), the file is parsed on your machine — no heavy PDF processing on our server.

Python (.py) and SQL (.sql) files are parsed with code-aware chunking — see Code-Aware Chunking below.

result = client.pipeline.chunk_file("./docs/manual.pdf")
print(f"Got {result.total_chunks} chunks from {result.source}")
print(f"Cost: ${result.credits_used:.4f}")

Use fetch_mode="server" for scanned/image-only PDFs that need OCR:

client = ScrapedatshiClient(api_key="sds_...", fetch_mode="server")
result = client.pipeline.chunk_file("./scanned_report.pdf")  # OCR included

Crawl a website

Crawls via sitemap or spider and chunks all pages. Large sites are automatically batched server-side — no manual pagination needed.

# Sitemap crawl (default) — reads sitemap.xml
result = client.pipeline.crawl("https://docs.example.com", max_pages=20)
print(f"Crawled {result.pages_crawled} pages → {result.total_chunks} chunks")

# Spider crawl — follows links, works on any site
result = client.pipeline.crawl(
    "https://example.com",
    crawl_mode="spider",
    max_pages=10,
    include_pattern="/docs/",
    exclude_pattern="/blog/",
)

# JS rendering — use a headless browser for JavaScript-heavy pages
# Can help with certain access restrictions ($0.0050/URL surcharge)
result = client.pipeline.crawl(
    "https://example.com",
    js_render=True,
    max_pages=10,
)

# Large sites (>200 pages) are auto-batched
if result.auto_batched:
    print(f"Auto-batched: {result.batches_processed} batches of {result.batch_size} pages")

Session Capture — Local Playwright (v0.12.4+)

For enterprise portals protected by Okta, Duo, or any SSO/MFA flow that blocks automated login, use capture_session() to authenticate manually in a real browser window and capture the full session state.

How it works

capture_session() runs entirely on your local machine using your IP address — the same IP your corporate SSO trusts. Once you log in and press Enter, the SDK captures all cookies and localStorage tokens. The crawl then runs on scrapedatshi's cloud infrastructure using the captured session, which is accepted because session tokens are IP-independent.

capture_session()                    crawl() / scrape()
─────────────────                    ──────────────────
Runs locally on                      Runs on Scrapedatshi's
your machine                         cloud infrastructure
using YOUR IP                        using our server IPs
        │                                     │
        ▼                                     ▼
Opens real Chrome                    Sends HTTP requests
on your desktop                      with the captured
        │                            session state
        ▼                                     │
You log in manually                          ▼
(MFA, Okta, etc.)                    Target site accepts
        │                            the session — tokens
        ▼                            are IP-independent
Captures storage_state
(cookies + localStorage)

Install

pip install scrapedatshi[auth]
playwright install chromium

Usage

from scrapedatshi.auth import capture_session
from scrapedatshi import ScrapedatshiClient

# Step 1: Opens a real browser — log in, then press Enter
state = capture_session(
    "https://internal.company.com/login",
    save_to="session.auth.json",   # optional — save for reuse (gitignored)
)

# Step 2: Crawl with the captured session
client = ScrapedatshiClient()
result = client.pipeline.crawl(
    "https://internal.company.com",
    storage_state=state,
    max_pages=20,
)

Save and reuse sessions

import json

# Save
state = capture_session("https://internal.company.com/login", save_to="session.auth.json")

# Load later — no need to log in again until the session expires
with open("session.auth.json") as f:
    state = json.load(f)

result = client.pipeline.crawl("https://internal.company.com", storage_state=state)

Browser choice

# Defaults to Chromium — also supports Firefox and WebKit
state = capture_session("https://internal.company.com/login", browser="firefox")

⚠ Security Warning

The generated session.auth.json contains live security keys capable of impersonating your user profile. Never commit your .auth.json files to Git repositories. Generated test sandboxes created using scrapedatshi init are automatically pre-configured with .gitignore filters tracking *.auth.json.

Authenticated Scraping (v0.10.0+)

For pages behind a login wall, pass your session cookies and/or custom headers to any fetch method. Credentials are only sent to URLs within the permitted domain scope — never leaked to external domains.

# Scrape a login-walled page
result = client.pipeline.chunk_url(
    "https://internal.company.com/wiki/api-docs",
    cookies={"session": "abc123", "csrf": "xyz"},
    headers={"Authorization": "Bearer eyJ..."},
)

# Authenticated sitemap crawl — cookies stay on your machine
result = client.pipeline.crawl(
    "https://internal.company.com",
    cookies={"session": "abc123"},
    headers={"Authorization": "Bearer eyJ..."},
    max_pages=20,
)

# Spider crawl with subdomain scope
# Also crawls wiki.company.com, docs.company.com, etc.
result = client.pipeline.crawl(
    "https://company.com",
    crawl_mode="spider",
    cookies={"session": "abc123"},
    allow_subdomains=True,   # safe: multi-part TLDs (.co.uk) handled correctly
    max_pages=30,
)

Security model:

• Cookies and headers are only sent to URLs within the permitted domain scope — never to external domains discovered during crawling
• allow_subdomains=False (default): only the exact hostname receives credentials
• allow_subdomains=True: credentials are shared with subdomains of the root domain. Multi-part TLDs (.co.uk, .com.br) are handled safely.
• Credentials are never forwarded to the scrapedatshi server — they stay on your machine

Full Pipeline — Embed + Inject

Scrape, embed, and inject directly into your vector database in one call. You bring your own embedding provider and vector DB keys (BYOK).

Sync a URL

result = client.pipeline.sync(
    url="https://docs.example.com",
    embedding_provider="openai",
    embedding_api_key="sk-...",
    embedding_model="text-embedding-3-small",
    vector_db="pinecone",
    vector_db_config={
        "api_key": "pc-...",
        "index_host": "https://my-index-abc123.svc.pinecone.io",
    },
    # hierarchical=True,      # optional: parent-child chunking for better retrieval
    # child_chunk_size=128,   # child chunk size when hierarchical=True
)
print(f"Upserted {result.vectors_upserted} vectors ({result.total_tokens} tokens)")
print(f"Cost: ${result.credits_used:.4f}")

Ingest a local file

result = client.pipeline.ingest(
    file_path="./docs/manual.pdf",
    embedding_provider="openai",
    embedding_api_key="sk-...",
    embedding_model="text-embedding-3-small",
    vector_db="qdrant",
    vector_db_config={
        "url": "https://your-cluster.qdrant.io",
        "collection_name": "documents",
        "api_key": "qdrant-key",
    },
    # hierarchical=True,      # optional: parent-child chunking for better retrieval
)
print(f"Ingested {result.chunks_created} chunks → {result.vectors_upserted} vectors")

Ingest scraped output (bulk import) — v0.11.0+

Bulk-ingest an entire folder of pre-scraped files into your vector database. Designed for output from web scrapers (Scrapy, Playwright, Apify, custom scripts). Supports all common file types including .md, .txt, .json, .yaml, .yml, .csv, .xlsx, .docx, .py, .sql, and more. JSON arrays are automatically detected and each item is extracted and ingested individually.

result = client.pipeline.ingest_scraped(
    folder_path="./scraped_output/",
    embedding_provider="openai",
    embedding_api_key="sk-...",
    embedding_model="text-embedding-3-small",
    vector_db="pinecone",
    vector_db_config={
        "api_key": "pc-...",
        "index_host": "https://my-index-abc123.svc.pinecone.io",
    },
)
print(f"Processed {result.files_processed} files → {result.vectors_upserted} vectors")
print(f"Failed: {result.files_failed} files")
print(f"Cost: ${result.credits_used:.4f}")
for err in result.errors:
    print(f"  ✗ {err['file']} — {err['error']}")

# Restrict to specific file types + add delay between files
result = client.pipeline.ingest_scraped(
    folder_path="./",
    file_extensions=[".json"],   # only process JSON files
    batch_delay=1.0,             # 1s pause between files (rate limit safety)
    embedding_provider="openai",
    embedding_api_key="sk-...",
    embedding_model="text-embedding-3-small",
    vector_db="pinecone",
    vector_db_config={"api_key": "pc-...", "index_host": "https://..."},
)

# Async version
result = await client.pipeline.ingest_scraped_async(
    folder_path="./docs/",
    embedding_provider="openai",
    embedding_api_key="sk-...",
    embedding_model="text-embedding-3-small",
    vector_db="qdrant",
    vector_db_config={"url": "https://...", "collection_name": "docs", "api_key": "..."},
)

IngestScrapedResult model:

result.files_processed      # int — number of files successfully ingested
result.files_failed         # int — number of files that failed
result.total_chunks         # int — total chunks created across all files
result.vectors_upserted     # int — total vectors upserted
result.credits_used         # float
result.credits_remaining    # float
result.errors               # list[dict] — [{"file": "...", "error": "..."}, ...]

AutoRAG — crawl entire site → embed → inject

result = client.pipeline.autorag(
    url="https://docs.example.com",
    max_pages=50,
    crawl_mode="sitemap",   # or "spider"
    embedding_provider="openai",
    embedding_api_key="sk-...",
    embedding_model="text-embedding-3-small",
    vector_db="pinecone",
    vector_db_config={"api_key": "pc-...", "index_host": "https://..."},
)
print(f"Crawled {result.pages_crawled} pages → {result.vectors_upserted} vectors")

# Large sites are auto-batched — no manual pagination needed
result = client.pipeline.autorag(
    url="https://large-docs-site.com",
    max_pages=800,  # processed as 4 batches of 200 pages each
    embedding_provider="openai",
    embedding_api_key="sk-...",
    embedding_model="text-embedding-3-small",
    vector_db="pinecone",
    vector_db_config={"api_key": "pc-...", "index_host": "https://..."},
)

Query Your Vector Database

Inspect a vector database (free)

Use this first to confirm the dimension and embedding model used during ingestion.

result = client.pipeline.inspect_vectordb(
    vector_db="pinecone",
    vector_db_config={
        "api_key": os.getenv("PINECONE_API_KEY"),
        "index_host": os.getenv("PINECONE_INDEX_HOST"),
    },
)
print(f"Dimension: {result.dimension}")
print(f"Vectors: {result.total_vector_count:,}")
print(f"Suggested models: {[m.label for m in result.suggested_models]}")

inspect_vectordb() is always free — no credits charged.

Query a vector database

result = client.pipeline.query_vectordb(
    query="How do I authenticate with the API?",
    embedding_provider="openai",
    embedding_api_key=os.getenv("OPENAI_API_KEY"),
    embedding_model="text-embedding-3-small",  # must match ingestion model
    vector_db="pinecone",
    vector_db_config={
        "api_key": os.getenv("PINECONE_API_KEY"),
        "index_host": os.getenv("PINECONE_INDEX_HOST"),
    },
    top_k=5,
)
print(f"Found {result.chunks_retrieved} results (cost: ${result.credits_used:.4f})")
for r in result.results:
    print(f"  [{r.score:.2f}] {r.text[:100]}...")
    # For hierarchical chunks, use parent_text as LLM context:
    if r.metadata.get("is_hierarchical") and r.metadata.get("parent_text"):
        print(f"  → LLM context: {r.metadata['parent_text'][:120]}...")

Billing: $0.0002 per chunk returned. Default top_k=5 → $0.001 per query.

Hybrid Search — Vector + BM25 Keyword (RRF)

Standard vector search is "single-hop" — it finds semantically similar chunks but can miss exact keyword matches (IDs, error codes, names). Hybrid search combines dense vector similarity with BM25 keyword search using Reciprocal Rank Fusion (RRF), enabling cross-referencing across separate documents.

result = client.pipeline.query_vectordb(
    query="Who manages Project Alpha?",
    embedding_provider="openai",
    embedding_api_key=os.getenv("OPENAI_API_KEY"),
    embedding_model="text-embedding-3-small",
    vector_db="lancedb",
    vector_db_config={"db_path": "./lancedb", "table_name": "docs"},
    top_k=5,
    hybrid_search=True,   # ← enable BM25 + vector + RRF
)

print(f"Hybrid search: {result.hybrid_search}")
for r in result.results:
    print(f"  [rrf={r.rrf_score:.4f}] sources={r.hybrid_sources} — {r.text[:80]}...")
    # r.hybrid_sources: ['vector'], ['keyword'], or ['vector', 'keyword']
    # Chunks in both lists get a combined RRF score boost

Provider support:

• LanceDB — native FTS index (best accuracy; requires FTS index on table)
• Qdrant — native MatchText payload filter
• Supabase — PostgreSQL ts_rank full-text search
• All others (Pinecone, Chroma, Weaviate, MongoDB) — client-side keyword scoring fallback

When to use hybrid search:

• Queries containing exact identifiers: IDs, error codes, product names, version numbers
• Multi-hop questions: "Who manages Project Alpha?" (requires linking ID 902 → Sarah Jenkins)
• Business data with structured fields that semantic search alone misses

RAG Chat — retrieve chunks and generate a grounded answer

result = client.pipeline.rag_chat(
    query="How do I authenticate with the API?",
    embedding_provider="openai",
    embedding_api_key=os.getenv("OPENAI_API_KEY"),
    embedding_model="text-embedding-3-small",
    vector_db="pinecone",
    vector_db_config={
        "api_key": os.getenv("PINECONE_API_KEY"),
        "index_host": os.getenv("PINECONE_INDEX_HOST"),
    },
    llm_provider="openai",
    llm_api_key=os.getenv("OPENAI_API_KEY"),
    llm_model="gpt-4o-mini",
    top_k=5,
)
print(result.answer)
print(f"Based on {result.chunks_retrieved} chunks (cost: ${result.credits_used:.4f})")
for source in result.sources:
    print(f"  [{source.score:.2f}] {source.text[:80]}...")

Billing: $0.0002 per chunk retrieved. LLM tokens are your own cost — scrapedatshi does not bill for LLM usage.

RAG Chat with hybrid search + query rewriting

result = client.pipeline.rag_chat(
    query="what about the second pricing tier?",
    embedding_provider="openai",
    embedding_api_key=os.getenv("OPENAI_API_KEY"),
    embedding_model="text-embedding-3-small",
    vector_db="pinecone",
    vector_db_config={
        "api_key": os.getenv("PINECONE_API_KEY"),
        "index_host": os.getenv("PINECONE_INDEX_HOST"),
    },
    llm_provider="openai",
    llm_api_key=os.getenv("OPENAI_API_KEY"),
    llm_model="gpt-4o-mini",
    top_k=5,
    hybrid_search=True,    # BM25 + vector + RRF
    query_rewrite=True,    # rewrite query before embedding (uses same LLM — no extra keys)
    conversation_history=[
        {"role": "user",      "content": "Tell me about the pricing tiers"},
        {"role": "assistant", "content": "There are three tiers: Basic, Pro, and Enterprise..."},
    ],
)
print(result.answer)
if result.rewritten_query:
    print(f"Searched for: {result.rewritten_query}")
if result.hybrid_search:
    print("Hybrid search was used (vector + BM25 + RRF)")

query_rewrite=True rewrites the raw query into a crisp, self-contained search query before embedding, using the same llm_provider/llm_api_key/llm_model you already provided for answer generation — no extra credentials needed. Resolves pronouns ("it", "that one", "the second") using conversation_history. Falls back to the original query on any error.

RagChatResult model:

result.query               # str — the original query string
result.answer              # str — LLM-generated grounded answer
result.embedding_provider  # str
result.embedding_model     # str
result.vector_db_provider  # str
result.llm_provider        # str
result.llm_model           # str
result.top_k_requested     # int
result.chunks_retrieved    # int
result.hybrid_search       # bool — True when hybrid search was used
result.rewritten_query     # str | None — the rewritten query (when query_rewrite=True)
result.sources             # list[QueryResult] — source chunks used to generate the answer
result.credits_used        # float
result.credits_remaining   # float
result.llm_error           # str | None — set if LLM call failed but chunks were retrieved

Query Rewriting for query_vectordb()

For retrieval-only use cases (no LLM answer generation), pass a query_rewrite config dict:

result = client.pipeline.query_vectordb(
    query="what about the second one?",
    embedding_provider="openai",
    embedding_api_key=os.getenv("OPENAI_API_KEY"),
    embedding_model="text-embedding-3-small",
    vector_db="pinecone",
    vector_db_config={
        "api_key": os.getenv("PINECONE_API_KEY"),
        "index_host": os.getenv("PINECONE_INDEX_HOST"),
    },
    top_k=5,
    hybrid_search=True,
    query_rewrite={
        "llm_provider": "openai",
        "llm_api_key": os.getenv("OPENAI_API_KEY"),
        "llm_model": "gpt-4o-mini",
        # Optional: prior turns for pronoun resolution
        "conversation_history": [
            {"role": "user",      "content": "Tell me about the refund policy"},
            {"role": "assistant", "content": "The refund window is 30 days..."},
        ],
    },
)
if result.rewritten_query:
    print(f"Searched for: {result.rewritten_query}")

The rewrite LLM does not need to match the embedding model — any provider (openai, anthropic, gemini) can be used regardless of what embedding model was used at ingestion time.

CLI — Quick Query

The scrapedatshi CLI also supports query and rag-chat commands for quick terminal searches:

# Semantic search
scrapedatshi query "how do I authenticate?" \
  --vector-db pinecone \
  --hybrid

# RAG chat answer
scrapedatshi rag-chat "how do I authenticate?" \
  --vector-db pinecone \
  --llm-provider openai \
  --llm-model gpt-4o-mini \
  --query-rewrite

# API keys are resolved from env vars automatically
# (OPENAI_API_KEY, PINECONE_API_KEY, PINECONE_INDEX_HOST, etc.)

Run scrapedatshi --help for the full list of options.

Schema Extraction

Extract structured data from any URL using your own LLM key. Define a schema and the API returns a typed JSON object.

Extract a single object

result = client.pipeline.extract(
    url="https://example.com/products/widget-pro",
    schema={
        "title": "string — the product name",
        "price": "number — the price in USD",
        "in_stock": "boolean — whether the item is in stock",
        "description": "string — the product description",
    },
    llm_provider="openai",
    llm_api_key="sk-...",
    llm_model="gpt-4o-mini",
)
print(result.extracted)
# → {"title": "Widget Pro", "price": 29.99, "in_stock": True, "description": "..."}
print(f"Cost: ${result.credits_used:.4f}")

Extract a list of items

Use extract_as_list=True for pages with multiple matching items (product listings, article feeds, search results):

result = client.pipeline.extract(
    url="https://example.com/products",
    schema={
        "title": "string — the product name",
        "price": "number — the price in USD",
    },
    llm_provider="openai",
    llm_api_key="sk-...",
    llm_model="gpt-4o-mini",
    extract_as_list=True,
)
print(f"Extracted {result.item_count} products")
for product in result.extracted:
    print(f"  {product['title']}: ${product['price']}")

Schema Extraction via Crawl

Crawl an entire domain and extract structured data from every page in a single call. Each page is processed independently — failed pages return an error object without aborting the batch. Only successfully extracted pages are billed.

result = client.pipeline.extract_crawl(
    url="https://example.com/products",
    schema={
        "title": "string — the product name",
        "price": "number — the price in USD",
        "in_stock": "boolean — whether the item is in stock",
    },
    llm_provider="openai",
    llm_api_key="sk-...",
    llm_model="gpt-4o-mini",
    max_pages=20,
    crawl_mode="sitemap",       # "sitemap" (default) or "spider"
    include_pattern="/products/",
)
print(f"Extracted {result.pages_extracted}/{result.pages_attempted} pages")
print(f"Cost: ${result.credits_used:.4f}")

for page in result.results:
    if page.ok:
        print(f"  {page.url}: {page.extracted}")
    else:
        print(f"  {page.url}: FAILED — {page.error}")

# Access only successful results
for page in result.successful_results:
    print(page.extracted["title"], page.extracted["price"])

Extract a list of items per page

Use extract_as_list=True when each crawled page contains multiple matching items (e.g. a category listing page with many products):

result = client.pipeline.extract_crawl(
    url="https://example.com/shop",
    schema={
        "title": "string — the product name",
        "price": "number — the price in USD",
    },
    llm_provider="openai",
    llm_api_key="sk-...",
    llm_model="gpt-4o-mini",
    max_pages=5,
    extract_as_list=True,   # each page returns a list of items instead of one object
)
for page in result.successful_results:
    print(f"{page.url}: {len(page.extracted)} items")
    for item in page.extracted:
        print(f"  {item['title']}: ${item['price']}")

Rate limiting with llm_rpm

Use llm_rpm to throttle LLM calls and avoid hitting provider rate limits. The server will space out requests to stay within the specified requests-per-minute budget.

result = client.pipeline.extract_crawl(
    url="https://pawsafe.com/collections/all",
    schema={
        "title":        "string — the product title",
        "price":        "number — price in USD",
        "in_stock":     "boolean — true if currently in stock",
        "review_count": "number — total number of customer reviews if visible",
        "rating":       "number — average star rating out of 5",
        "image_url":    "string — main image link",
    },
    llm_provider="gemini",
    llm_api_key="AIza...",
    llm_model="gemini-2.5-flash",
    max_pages=50,
    crawl_mode="sitemap",
    include_pattern="/products/",
    llm_rpm=10,   # ← throttle to 10 calls/min for Gemini Tier-1
)
print(f"Extracted {result.pages_extracted}/{result.pages_attempted} pages")

When llm_rpm is omitted, the server uses a short jittered politeness delay (0.4–0.9 s) between pages — suitable for providers with high rate limits.

Billing: $0.0020 + $0.0030 + (N_fields × $0.0001) per successfully extracted page.

Contextual Retrieval (RAG 2.0)

For each chunk, an LLM generates a unique context string describing the document identity, section identity, and specific entities in that chunk. This context is prepended to the chunk text before embedding, boosting retrieval accuracy by 35–50%.

Pricing: $0.0010 per chunk successfully enriched.

result = client.pipeline.chunk_url(
    "https://docs.example.com",
    contextual_retrieval=True,
    llm_provider="openai",
    llm_api_key="sk-...",
    llm_model="gpt-4o-mini",
)

for chunk in result.chunks:
    print(chunk.context)        # LLM-generated context for this specific chunk
    print(chunk.original_text)  # Raw chunk text before enrichment
    print(chunk.content)        # Combined: "Context: ...\n\n{original_text}"

if result.contextual_retrieval_error:
    print(f"CR warning: {result.contextual_retrieval_error}")

Available on all pipeline methods: chunk_url(), chunk_file(), crawl(), sync(), ingest().

Supported Providers

Discover all supported providers programmatically:

from scrapedatshi.providers import (
    EMBEDDING_PROVIDERS,
    VECTOR_DB_PROVIDERS,
    LLM_PROVIDERS,
)

for key, info in EMBEDDING_PROVIDERS.items():
    print(f"{key}: {info['label']}")

print(VECTOR_DB_PROVIDERS["pinecone"]["required_fields"])
# → ["api_key", "index_host"]

Embedding Providers

Vector Database Providers

LLM Providers (for Contextual Retrieval & Schema Extraction)

Note: The document processing window applies to /v1/extract and /v1/extract-crawl only — it is a scrapedatshi server-side limit on how much page text is sent to the LLM, not the model's actual token limit. Use an advanced model for long-form pages.

Local Providers

Ollama (Local Embedding)

Ollama lets you run embedding models locally — no API key required. Because the scrapedatshi API server needs to reach your Ollama instance, you must expose it publicly using ngrok before use.

ollama pull nomic-embed-text
ngrok http 11434
# → Forwarding: https://abc123.ngrok-free.app → localhost:11434

result = client.pipeline.sync(
    url="https://docs.example.com",
    embedding_provider="ollama",
    embedding_api_key="",
    embedding_model="nomic-embed-text",
    embedding_endpoint="https://abc123.ngrok-free.app",
    vector_db="chroma",
    vector_db_config={"collection_name": "docs"},
)

ChromaDB (Local Vector DB)

pip install chromadb
chroma run --path ./chroma_data

result = client.pipeline.sync(
    url="https://docs.example.com",
    embedding_provider="openai",
    embedding_api_key="sk-...",
    embedding_model="text-embedding-3-small",
    vector_db="chroma",
    vector_db_config={
        "collection_name": "my_docs",
        "host": "localhost",
        "port": 8000,
    },
)

LanceDB (Local Vector DB)

result = client.pipeline.sync(
    url="https://docs.example.com",
    embedding_provider="openai",
    embedding_api_key="sk-...",
    embedding_model="text-embedding-3-small",
    vector_db="lancedb",
    vector_db_config={
        "db_path": "./lancedb",
        "table_name": "documents",
    },
)

Async Support

All methods have an _async variant for use with asyncio.

import asyncio
from scrapedatshi import ScrapedatshiClient

async def main():
    async with ScrapedatshiClient(api_key="sds_...") as client:
        result = await client.pipeline.chunk_url_async("https://docs.example.com")
        print(f"Got {result.total_chunks} chunks — cost ${result.credits_used:.4f}")

asyncio.run(main())

Parallel processing with asyncio.gather

async def main():
    async with ScrapedatshiClient(api_key="sds_...") as client:
        urls = [
            "https://docs.example.com/page1",
            "https://docs.example.com/page2",
            "https://docs.example.com/page3",
        ]
        results = await asyncio.gather(
            *[client.pipeline.chunk_url_async(url) for url in urls]
        )
        total = sum(r.total_chunks for r in results)
        total_cost = sum(r.credits_used for r in results)
        print(f"Processed {len(urls)} URLs → {total} chunks — total cost ${total_cost:.4f}")

Response Models

All methods return typed Pydantic models with full IDE autocomplete support. Every response includes credits_used and credits_remaining for programmatic spend tracking.

ChunkResult

result.chunks                  # list[Chunk]
result.total_chunks            # int
result.source                  # str
result.selectors_found         # list[str] — CSS selectors for detected content sections
result.hierarchical            # bool — True when hierarchical (parent-child) chunking was used
result.contextual_retrieval_used  # bool
result.content_truncated       # bool — True if content exceeded ~75,000 words
result.credits_used            # float
result.credits_remaining       # float

Chunk

chunk.content              # str — the chunk text (child chunk when hierarchical=True)
chunk.token_estimate       # int — estimated token count
chunk.original_text        # str | None — raw text before CR enrichment
chunk.context              # str | None — LLM-generated per-chunk context
chunk.parent_text          # str | None — larger parent chunk text (hierarchical mode only)
chunk.parent_index         # int | None — which parent this child belongs to
chunk.is_hierarchical      # bool — True when produced by hierarchical chunking
chunk.metadata             # dict — source URL, page number, etc.
chunk.code_metadata        # dict | None — set for .py/.sql files (see Code-Aware Chunking)

CrawlChunkResult

result.chunks              # list[Chunk]
result.total_chunks        # int
result.pages_crawled       # int
result.source_url          # str
result.credits_used        # float
result.credits_remaining   # float

SyncResult / IngestResult

result.status              # "success" | "partial" | "error"
result.chunks_created      # int
result.vectors_upserted    # int
result.total_tokens        # int
result.embedding_provider  # str
result.vector_db_provider  # str
result.hierarchical        # bool — True when hierarchical chunking was used
result.credits_used        # float
result.credits_remaining   # float

QueryVectorDBResult

result.query               # str — the original query string
result.chunks_retrieved    # int
result.hybrid_search       # bool — True when hybrid (vector + BM25 + RRF) search was used
result.rewritten_query     # str | None — the rewritten query used for search (when query_rewrite was enabled)
result.results             # list[QueryResult]
result.credits_used        # float
result.credits_remaining   # float

QueryResult

r.text             # str — chunk text (child chunk when hierarchical)
r.score            # float — similarity score (0–1)
r.metadata         # dict — url, chunk_index, parent_text, is_hierarchical, etc.
r.rrf_score        # float | None — RRF score when hybrid_search=True
r.hybrid_sources   # list[str] | None — ['vector'], ['keyword'], or ['vector', 'keyword']

# For hierarchical chunks — use parent_text as LLM context:
if r.metadata.get("is_hierarchical"):
    llm_context = r.metadata.get("parent_text", r.text)

IngestScrapedResult

result.files_processed      # int
result.files_failed         # int
result.total_chunks         # int
result.vectors_upserted     # int
result.embedding_provider   # str
result.vector_db_provider   # str
result.credits_used         # float
result.credits_remaining    # float
result.errors               # list[dict] — [{"file": "...", "error": "..."}, ...]

ExtractResult

result.extracted           # dict | list[dict]
result.field_count         # int
result.item_count          # int | None — list mode only
result.is_list             # bool
result.url                 # str
result.llm_provider        # str
result.llm_model           # str
result.schema_fields       # list[str]
result.js_render           # bool
result.content_warning     # str | None
result.credits_used        # float
result.credits_remaining   # float

ExtractCrawlResult

result.results             # list[ExtractCrawlPageResult]
result.pages_extracted     # int
result.pages_failed        # int
result.pages_attempted     # int
result.successful_results  # list[ExtractCrawlPageResult]
result.failed_results      # list[ExtractCrawlPageResult]
result.credits_used        # float
result.credits_remaining   # float

Each ExtractCrawlPageResult:

page.url        # str
page.status     # "ok" | "error"
page.extracted  # dict | list[dict] | None
page.error      # str | None
page.ok         # bool

Pricing

scrapedatshi uses a pay-per-use credit wallet — no subscriptions, no monthly fees. Credits are deducted after each successful API call. Failed requests are never charged.

Top up your balance at scrapedatshi.com/portal/billing.

Hard Caps

Per-request hard caps protect server stability and apply to all accounts:

Content exceeding the size limit is automatically truncated — check result.content_truncated to detect this.

Error Handling

from scrapedatshi.exceptions import (
    AuthError,                 # Invalid or missing API key (401/403)
    InsufficientCreditsError,  # Balance too low (402)
    RateLimitError,            # Rate limit exceeded (429)
    ValidationError,           # Bad request payload (422)
    ServerBusyError,           # Server at capacity — retry after e.retry_after seconds (503)
    ServerError,               # API server error (5xx)
    TimeoutError,              # Request timed out
    ScrapedatshiError          # Base exception — catch-all
)

try:
    result = client.pipeline.sync(
        url="https://docs.example.com",
        embedding_provider="openai",
        embedding_api_key="sk-...",
        vector_db="pinecone",
        vector_db_config={"api_key": "pc-...", "index_host": "https://..."},
    )
except InsufficientCreditsError:
    print("Balance too low — top up at scrapedatshi.com/portal/billing")
except RateLimitError as e:
    print(f"Rate limit hit: {e.message}")
except ScrapedatshiError as e:
    print(f"API error {e.status_code}: {e.message}")

Handling ServerBusyError (503)

import time
from scrapedatshi.exceptions import ServerBusyError

try:
    result = client.pipeline.extract_crawl(...)
except ServerBusyError as e:
    wait = e.retry_after or 30
    print(f"Server busy — retrying in {wait}s")
    time.sleep(wait)
    # retry the request...

Troubleshooting

Contextual Retrieval fails — deprecated or unavailable model

LLM providers periodically deprecate older models. When contextual retrieval fails due to a deprecated model, the SDK will emit a UserWarning automatically. Check the error programmatically:

result = client.pipeline.chunk_url(
    "https://example.com",
    contextual_retrieval=True,
    llm_provider="gemini",
    llm_api_key="AIza...",
    llm_model="models/gemini-2.5-flash",  # use a current model
)
if result.contextual_retrieval_error:
    print(f"CR warning: {result.contextual_retrieval_error}")

Provider model & deprecation pages:

• OpenAI: platform.openai.com/docs/deprecations
• Anthropic: docs.anthropic.com/en/docs/about-claude/models
• Google Gemini: ai.google.dev/gemini-api/docs/models
• Cohere: docs.cohere.com/docs/models
• Mistral: docs.mistral.ai/getting-started/models/
• Voyage AI: docs.voyageai.com/docs/embeddings

Contextual Retrieval fails — quota exceeded

Your LLM provider API key has no remaining credits. Note that scrapedatshi credits and LLM provider credits are separate — you need both.

Suppressing contextual retrieval warnings

import warnings

with warnings.catch_warnings():
    warnings.simplefilter("ignore", UserWarning)
    result = client.pipeline.chunk_url(
        "https://example.com",
        contextual_retrieval=True,
        ...
    )

Code-Aware Chunking

When ingesting .py or .sql files via chunk_file(), ingest(), or ingest_folder(), the SDK uses AST/statement-aware splitting instead of treating the file as a plain text blob. This produces semantically meaningful chunks that align with logical code boundaries.

Python (.py) — AST-aware

Uses Python's stdlib ast module to extract top-level classes and functions as individual units. Each unit includes the relevant import statements prepended as context.

result = client.pipeline.chunk_file("./src/models.py")
# → Each class and top-level function becomes its own chunk
# → Imports are prepended to each unit for context

Fallback: If the file has a syntax error, the whole file is treated as a single plain-text chunk.

What gets split:

• Top-level class definitions (including all their methods)
• Top-level def and async def functions
• Files with no top-level definitions (e.g. pure scripts) → single chunk

SQL (.sql) — Statement-aware

Uses regex to detect the start of each major SQL statement and groups lines between boundaries.

result = client.pipeline.chunk_file("./schema.sql")
# → Each CREATE TABLE, CREATE FUNCTION, INSERT INTO, SELECT, etc. becomes its own chunk

Recognised statement types: CREATE TABLE, CREATE VIEW, CREATE PROCEDURE, CREATE FUNCTION, CREATE TRIGGER, CREATE INDEX, ALTER TABLE, DROP TABLE, INSERT INTO, UPDATE, DELETE FROM, SELECT, WITH, MERGE, TRUNCATE, GRANT, REVOKE.

Fallback: If no statement boundaries are found, the whole file is treated as a single chunk.

Fragment URLs

Each logical unit gets a unique fragment appended to its source URL so you can trace chunks back to their origin:

file://models.py#UserModel          ← Python class
file://models.py#get_user           ← Python function
file://schema.sql#create_table_0    ← first CREATE TABLE
file://schema.sql#create_table_1    ← second CREATE TABLE

Development

git clone https://github.com/scrapedatshi/scrapedatshi-py
cd scrapedatshi-py
pip install -e ".[dev]"
pytest

License

MIT — see LICENSE.