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_...")

# Chunk a URL to JSON (no embedding required)
result = client.pipeline.chunk_url("https://docs.example.com")

print(f"Got {result.total_chunks} chunks")
print(f"Cost: ${result.credits_used:.4f} | Remaining: ${result.credits_remaining:.4f}")
for chunk in result.chunks:
    print(chunk.content[:80])

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. New accounts receive $1.00 free credits — no credit card required.

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.

Fetch Mode Pricing

Starting in v0.8.0, the SDK uses local-fetch mode by default — your machine fetches the URL using your own IP address and submits the HTML to our server for processing. This is cheaper and keeps your IP off our server.

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

Fetch Mode

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

fetch_mode="local" (default — v0.8.0+)

The SDK fetches the URL on your machine using your IP address, then submits the raw HTML to our server for processing. This is the default and recommended mode.

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

# Default — local fetch (your IP)
client = ScrapedatshiClient(api_key="sds_...")
result = client.pipeline.chunk_url("https://docs.example.com")

fetch_mode="server"

Our server fetches the URL. Use this if you are behind a corporate firewall, need server-managed IP rotation, or are running in an environment without outbound HTTP access.

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

# Server fetch — our server fetches the URL
client = ScrapedatshiClient(api_key="sds_...", fetch_mode="server")
result = client.pipeline.chunk_url("https://docs.example.com")

Note: The portal no-code tools always use server fetch and are billed at the server fetch rate.

Authenticated Scraping (v0.10.0+)

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

Single URL

# Pass your browser session cookie
result = client.pipeline.chunk_url(
    "https://internal.company.com/wiki/api-docs",
    cookies={"session": "abc123", "csrf": "xyz"},
    headers={"Authorization": "Bearer eyJ..."},
)

# Full pipeline with authentication
result = client.pipeline.sync(
    url="https://internal.company.com/wiki/api-docs",
    cookies={"session": "abc123"},
    embedding_provider="openai",
    embedding_api_key="sk-...",
    vector_db="pinecone",
    vector_db_config={"api_key": "pc-...", "index_host": "https://..."},
)

Crawl with Authentication

# 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 (e.g. wiki.company.com when root is company.com). Multi-part TLDs (.co.uk, .com.br) are handled safely.
• Credentials are never forwarded to the scrapedatshi server — they stay on your machine

Pipeline Methods

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")

# result.chunks              → list[Chunk]
# result.total_chunks        → int
# result.source              → str (the URL)
# result.credits_used        → float
# result.credits_remaining   → float
# result.content_truncated   → bool (True if content exceeded ~75,000 words)

Chunk a PDF URL

Pass any PDF URL directly — S3 links, CDN URLs, direct .pdf links — and the API automatically detects and extracts text using pdfplumber (text-layer) with RapidOCR fallback for scanned documents. No special parameters needed.

Chunk a local file (PDF, MD, TXT, YAML, JSON)

In local-fetch mode (default), the file is parsed on your machine using your own CPU — no heavy PDF processing on our server. The extracted text is sent to our server for chunking only. PDF support is included with the standard install.

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}")  # billed at local rate ($0.0020)

Fetch mode for files:

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

Query Your Vector Database

After ingesting data, query it using natural language. Use inspect_vectordb() first to confirm the correct embedding model.

Inspect a vector database (free)

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]}")
# → Suggested models: ['OpenAI text-embedding-3-small', 'OpenAI text-embedding-ada-002 (legacy)']

Query a vector database

# Confirm the model from inspect_vectordb first, then query
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]}...")

Billing: $0.0002 per chunk returned. Default top_k=5 → $0.001 per query. inspect_vectordb() is always free.

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",  # must match ingestion model
    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 (same as query_vectordb()). LLM tokens are your own cost — scrapedatshi does not bill for LLM usage.

# Direct S3 PDF link — automatically detected and extracted
result = client.pipeline.chunk_url(
    "https://my-bucket.s3.amazonaws.com/reports/annual-report-2024.pdf"
)

# CDN-hosted PDF without .pdf extension — detected via Content-Type header
result = client.pipeline.chunk_url(
    "https://cdn.example.com/documents/abc123"
)

print(f"Got {result.total_chunks} chunks from PDF")

Chunk a URL with JS rendering

For JavaScript-heavy pages and SPAs that require a browser to render:

result = client.pipeline.chunk_url(
    "https://spa.example.com/dashboard",
    js_render=True,
)

Chunk a local file

Supports PDF, MD, TXT, YAML, YML, and JSON.

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}")

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://example.com", max_pages=10)

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

print(f"Crawled {result.pages_crawled} pages → {result.total_chunks} chunks")
print(f"Cost: ${result.credits_used:.4f}")

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

Full Pipeline — Embed + Inject

Scrape, embed, and inject directly into your vector database in one call.

Sync a URL

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://my-index-abc123.svc.pinecone.io",
    },
)

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-...",
    vector_db="qdrant",
    vector_db_config={
        "url": "https://your-cluster.qdrant.io",
        "collection_name": "documents",
        "api_key": "qdrant-key",  # optional for local Qdrant
    },
)

Ingest a folder (bulk import) — v0.11.0+

Bulk-ingest an entire folder of pre-scraped files into your vector database. Supports .md, .txt, .json, .yaml, and .yml. Automatically handles Scrapy JSON array exports — each item in the array is extracted and ingested individually. Includes exponential backoff on rate limits and a configurable per-file delay.

# Ingest all supported files in a folder
result = client.pipeline.ingest_folder(
    folder_path="./scrapy_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']}")

# Scrapy JSON dump — each item in the array is ingested individually.
# Items with 'text', 'content', 'html', 'body', 'markdown', or 'description'
# fields are automatically detected and extracted.
result = client.pipeline.ingest_folder(
    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_folder_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://your-cluster.qdrant.io",
        "collection_name": "docs",
        "api_key": "qdrant-key",
    },
)

IngestFolderResult 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.embedding_provider   # str
result.vector_db_provider   # str
result.credits_used         # float
result.credits_remaining    # float
result.errors               # list[dict] — [{"file": "...", "error": "..."}, ...]

Billing: $0.0020 / file · $0.0005 / chunk · $0.0030 / chunk injected. Only successfully processed files are billed.

Schema Extraction

Extract structured data from any URL using your own LLM key. Define a schema and the API returns a typed JSON object — or a list of objects for pages with multiple items.

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-...",
)

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-...",
    extract_as_list=True,
)

print(f"Extracted {result.item_count} products")
for product in result.extracted:
    print(f"  {product['title']}: ${product['price']}")

Extract from a JS-rendered page

result = client.pipeline.extract(
    url="https://spa.example.com/data",
    schema={"value": "string — the data value"},
    llm_provider="anthropic",
    llm_api_key="sk-ant-...",
    js_render=True,
)

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 (only charged for chunks where CR succeeded).

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

# Each chunk now has per-chunk context fields
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}"

# Check if CR partially failed (chunks still returned without context)
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,
)

# List all embedding providers
for key, info in EMBEDDING_PROVIDERS.items():
    print(f"{key}: {info['label']} (requires_api_key={info['requires_api_key']})")
    print(f"  {info['notes']}")

# Check required fields for a vector DB
print(VECTOR_DB_PROVIDERS["pinecone"]["required_fields"])
# → ["api_key", "index_host"]

# List LLM providers (for CR and schema extraction)
for key, info in LLM_PROVIDERS.items():
    print(f"{key}: {info['label']}")
    print(f"  {info['notes']}")

Embedding Providers

Embedding providers use embedding-specific models to convert text into vectors. Check your provider's documentation for available models.

Vector Database Providers

LLM Providers (for Contextual Retrieval & Schema Extraction)

LLM providers use chat/completion models — different from embedding models. A model name is always required; no default is applied. Check your provider's documentation for models available on your API key.

Document processing window (Schema Extraction only): This cap applies to /v1/extract and /v1/extract-crawl — it is a scrapedatshi server-side limit on how much page text is sent to the LLM for schema extraction, not the model's actual token limit. Standard models (names containing "mini", "flash", "haiku", "lite", or "nano") receive up to 8,000 characters; all other models receive up to 30,000 characters. Use an advanced model for long-form pages (documentation, legal docs, research papers) to ensure the full page is considered.

Note: This limit does not apply to Contextual Retrieval. CR uses a separate fixed document preview window and is not affected by model tier.

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 (or a similar tunnel) before use.

Setup:

# 1. Start Ollama and pull an embedding model
ollama pull nomic-embed-text

# 2. Expose it publicly with ngrok
ngrok http 11434
# → Forwarding: https://abc123.ngrok-free.app → localhost:11434

Usage:

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

Important: The embedding_endpoint must be the public ngrok HTTPS URL, not localhost. The API server cannot reach your local machine directly.

ChromaDB (Local Vector DB)

ChromaDB stores vectors as files on your local machine. The ChromaDB HTTP server must be running before you call the API.

pip install chromadb
chroma run --path ./chroma_data
# → ChromaDB running at http://localhost:8000

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",   # optional, default: localhost
        "port": 8000,          # optional, default: 8000
    },
)

LanceDB (Local Vector DB)

LanceDB stores vectors as files on your local filesystem — no server required.

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",      # local directory path
        "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} 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.contextual_retrieval_used  # bool
result.content_truncated       # bool — True if content exceeded ~75,000 words
result.credits_used            # float — credits deducted for this request
result.credits_remaining       # float — account balance after this request

Chunk

chunk.content              # str — the chunk text (combined "Context: ...\n\n{original_text}" when CR used)
chunk.token_estimate       # int — estimated token count
chunk.original_text        # str | None — raw text before CR enrichment (only set when CR succeeded)
chunk.context              # str | None — LLM-generated per-chunk context (only set when CR succeeded)
chunk.metadata             # dict — source URL, page number, etc.

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.credits_used        # float
result.credits_remaining   # float

ExtractResult

result.extracted           # dict | list[dict] — the extracted data
result.field_count         # int — number of schema fields
result.item_count          # int | None — number of items (list mode only)
result.is_list             # bool — True if extracted is a list
result.url                 # str — the URL that was scraped
result.llm_provider        # str
result.llm_model           # str
result.schema_fields       # list[str] — field names from your schema
result.js_render           # bool — whether JS rendering was used
result.content_warning     # str | None — warning if content may be incomplete
result.credits_used        # float
result.credits_remaining   # float

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-...",
    max_pages=20,
    include_pattern="/products/",
)

print(f"Extracted {result.pages_extracted}/{result.pages_attempted} pages")
print(f"Cost: ${result.credits_used:.4f} | Remaining: ${result.credits_remaining:.4f}")

# Iterate all results
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"])

Billing: $0.0020 + $0.0030 + (N_fields × $0.0001) per successfully extracted page. Example: 20 pages × 3 fields = 20 × $0.0053 = $0.106

Spider crawl mode

result = client.pipeline.extract_crawl(
    url="https://example.com",
    schema={"title": "string — the page title", "summary": "string — a brief summary"},
    llm_provider="anthropic",
    llm_api_key="sk-ant-...",
    crawl_mode="spider",
    max_pages=10,
)

ExtractCrawlResult model

result.results             # list[ExtractCrawlPageResult] — per-page results
result.pages_extracted     # int — successfully extracted
result.pages_failed        # int — failed (not billed)
result.pages_attempted     # int — total attempted
result.pages_discovered    # int — total URLs found in sitemap/spider
result.successful_results  # list[ExtractCrawlPageResult] — only ok pages
result.failed_results      # list[ExtractCrawlPageResult] — only failed pages
result.job_id              # str | None — persistent job ID
result.credits_used        # float
result.credits_remaining   # float

Each ExtractCrawlPageResult:

page.url        # str — the URL scraped
page.status     # "ok" | "error"
page.extracted  # dict | list[dict] | None — extracted data (None on error)
page.error      # str | None — error message (None on success)
page.ok         # bool — True if status == "ok"

Error Handling

from scrapedatshi.exceptions import (
    AuthError,              # Invalid or missing API key (401/403)
    InsufficientCreditsError,  # Balance too low — top up at portal/billing (402)
    RateLimitError,         # Per-request hard cap or 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)

Large crawl jobs use a server-side queue. When the queue is full, the API returns HTTP 503 with a Retry-After header. The SDK surfaces this as ServerBusyError with a retry_after attribute:

import time
from scrapedatshi.exceptions import ServerBusyError

try:
    result = client.pipeline.extract_crawl(
        url="https://example.com",
        schema={"title": "string — the page title"},
        llm_provider="openai",
        llm_api_key="sk-...",
        max_pages=50,
    )
except ServerBusyError as e:
    wait = e.retry_after or 30  # seconds to wait (from Retry-After header)
    print(f"Server busy — retrying in {wait}s")
    time.sleep(wait)
    # retry the request...

Auto-Batching for Large Sites

When a crawl job exceeds the per-batch page cap (200 pages), the API automatically splits it into sequential batches and processes them all in a single API call. You don't need to do anything — just submit the job and it returns when all batches are complete.

# This 800-page site will be processed as 4 batches of 200 pages each
result = client.pipeline.autorag(
    url="https://large-docs-site.com",
    max_pages=800,
    embedding_provider="openai",
    embedding_api_key="sk-...",
    vector_db="pinecone",
    vector_db_config={"api_key": "pc-...", "index_host": "https://..."},
)

print(f"Crawled {result.pages_crawled} pages → {result.vectors_upserted} vectors")
if result.auto_batched:
    print(f"Processed in {result.batches_processed} batches of {result.batch_size} pages each")

Auto-batching applies to: crawl(), autorag(), and the underlying /v1/crawl, /v1/autorag, /v1/crawl-chunk endpoints.

Hard Caps

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

Sitemap crawl (crawl_mode="sitemap"): Reads sitemap.xml to discover URLs. Jobs exceeding 200 pages are automatically batched — no manual pagination needed.

Spider crawl (crawl_mode="spider"): Follows <a href> links via BFS. More compute-intensive — start small and increase as needed. Also supports auto-batching.

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

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:

UserWarning: scrapedatshi contextual retrieval warning: The model 'models/gemini-2.0-flash'
is no longer available from Google Gemini. Please select a current model.
Check available models: https://ai.google.dev/gemini-api/docs/models

You can also 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 (open in new tab):

• 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. The contextual_retrieval_error field will contain an actionable message pointing to your provider's billing page. Note that scrapedatshi credits and LLM provider credits are separate — you need both.

Suppressing contextual retrieval warnings

If you handle contextual_retrieval_error programmatically and don't want the UserWarning, suppress it with Python's standard warnings module:

import warnings

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

Development

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

License

MIT — see LICENSE.