Skip to main content

DivParser Python SDK

A Python SDK for DivParser - AI-powered web scraping and HTML parsing.

Features

  • Web Scraping: Extract structured data from web pages
  • Fetch-then-Extract: Fetch a page first, decide on the extraction schema afterwards
  • HTML Parsing: Parse raw HTML content directly
  • Schedules: Create and manage recurring scrapes (cron or interval cadence)
  • Async Job Handling: Non-blocking job submission with status polling
  • Pagination Support: Scrape multiple URLs in a single batch
  • Export: Save results as JSON, CSV, or XLSX, and fetched HTML as-is or converted to Markdown
  • Delivery: Deliver results directly to a connected S3/Google Drive/Dropbox Storage
  • Simple API: Pythonic interface to the DivParser REST API

Installation

pip install divparser

Or if using uv:

uv pip install divparser

Quick Start

Setup

from divparser import DivParser

# Initialize the client with your API key
client = DivParser(api_key="your_api_key_here")

Get your API key from DivParser Console.

Scraping a Web Page

# Scrape a single page and wait for results
result = client.scrape_and_parse(
    url="https://example.com/products",
    schema="Extract product name, price, and rating from each item"
)

# Access the extracted data
for item in result["results"][0]["data"]:
    print(item)

Parsing HTML Content

# Parse HTML content directly
html_content = "<html><body><h1>Title</h1><p>Content</p></body></html>"

result = client.parse_and_wait(
    html=html_content,
    schema="Extract all headings and paragraphs"
)

# Get the parsed data
data = result["results"][0]["data"]
print(data)

Paginated Scraping

# Scrape multiple URLs
urls = [
    "https://example.com/page/1",
    "https://example.com/page/2",
    "https://example.com/page/3"
]

result = client.scrape_paginated(
    urls=urls,
    schema="Extract product name and price",
    wait=True
)

# Combine results from all pages
from divparser.utils import flatten_results
all_items = flatten_results(result["results"])

Fetch First, Extract Later

# Fetch the page now, decide on a schema afterwards
fetch_result = client.fetch(url="https://example.com/products")
client.wait_for_completion(fetch_result["jobId"])

extracted = client.extract_fetch(
    scrape_id=fetch_result["scrapeId"],
    schema="Extract product name and price"
)

Recurring Schedules

schedule = client.create_schedule(
    name="Daily product check",
    project_id="YOUR_PROJECT_ID",
    schedule={"type": "cron", "pattern": "0 9 * * *"},
    scrape={
        "url": "https://example.com/products",
        "schema": "Extract product name, price, and availability"
    }
)

# List the scrapes this schedule has generated so far
runs = client.list_schedule_runs(schedule["scheduleId"])

# Pause it, resume it, or delete it entirely
client.pause_schedule(schedule["scheduleId"])
client.resume_schedule(schedule["scheduleId"])
client.delete_schedule(schedule["scheduleId"])

Saving Results to Disk

from divparser import save_scrape_as, save_fetch_as

# Save a scrape/parse result as CSV, JSON, or XLSX — usable rows (SUCCESS
# and REQUIRES_ATTENTION) are combined into one flat export automatically.
result = client.get_scrape(scrape_id)
save_scrape_as(result["results"], "csv", "products.csv")

# Save a fetch's raw HTML, or convert it to Markdown first.
fetch_html = client.get_fetch_html(scrape_id)
save_fetch_as(fetch_html["html"], "markdown", "page.md")

API Reference

Scraping

scrape(url, schema, name=None, page_type="LISTING", wait=False, timeout=300, delivery_config=None)

Create a scrape job for a single URL.

Parameters:

  • url (str): Target page URL
  • schema (str): Extraction instructions (plain English or Nestlang)
  • name (str, optional): Friendly label for this scrape
  • page_type (str): "LISTING" (default) or "DETAIL"
  • wait (bool): Wait for completion before returning
  • timeout (int): Max seconds to wait (only if wait=True)
  • delivery_config (dict, optional): See Delivering Results to a Storage

Returns: Dictionary with scrapeId, jobId, and optionally results

scrape_paginated(urls, schema, name=None, page_type="LISTING", wait=False, timeout=300, delivery_config=None)

Create a scrape job for multiple URLs.

Parameters:

  • urls (List[str]): Array of URLs to scrape
  • schema (str): Extraction instructions
  • name (str, optional): Friendly label
  • page_type (str): "LISTING" or "DETAIL"
  • wait (bool): Wait for completion
  • timeout (int): Max seconds to wait
  • delivery_config (dict, optional): See Delivering Results to a Storage

Returns: Dictionary with scrapeId, jobId, and optionally results

list_scrapes(limit=20, cursor=None)

List all scrapes for the authenticated user.

Returns: Dictionary with list of scrapes and pagination info

get_scrape(scrape_id)

Retrieve a scrape and its results by ID.

Parameters:

  • scrape_id (str): The scrapeId from creation

Returns: Dictionary with scrape details and results

Parsing

parse(html, schema, name=None, wait=False, timeout=300)

Submit raw HTML for structured extraction.

Parameters:

  • html (str): Full HTML content to parse
  • schema (str): Extraction instructions
  • name (str, optional): Friendly label
  • wait (bool): Wait for completion
  • timeout (int): Max seconds to wait

Returns: Dictionary with scrapeId, jobId, and optionally results

get_parse(parse_id)

Retrieve results for a completed parse job.

Parameters:

  • parse_id (str): The scrapeId from parse creation

Returns: Dictionary with parse details and results

Fetching

fetch(url, name=None, project_id=None, proxy_mode=None, delivery_config=None)

Fetch a URL only, with no extraction.

Parameters:

  • url (str): Target page URL
  • name (str, optional): Friendly label
  • project_id (str, optional): Project to attach this fetch to
  • proxy_mode (str, optional): "residential", "unblocker", or "http"
  • delivery_config (dict, optional): See Delivering Results to a Storage — fetch content is raw HTML/Markdown, never tabular

Returns: Dictionary with scrapeId, jobId, and message

Note: this endpoint reports every failure as a 402 (not just insufficient credits) — check the response's message, not just the status code, to tell failures apart.

get_fetch_html(scrape_id)

Retrieve the raw HTML for a fetched scrape. Only meaningful for a scrape created via fetch() — get_scrape() never includes this (it's not an extraction result).

Parameters:

  • scrape_id (str): The scrapeId returned from fetch()

Returns: Dictionary with id and html

extract_fetch(scrape_id, schema, delivery_config=None)

Attach a schema to a previously-fetched scrape and run instant (AI) extraction against its already-stored HTML.

Parameters:

  • scrape_id (str): The scrapeId returned from fetch()
  • schema (str): Extraction instructions
  • delivery_config (dict, optional): See Delivering Results to a Storage — the extracted result is tabular, so the full destination set applies

Returns: Dictionary with scrapeId, jobId, and message

Scheduling

create_schedule(name, project_id, schedule, scrape, iterations=None, delivery_config=None)

Create a recurring schedule that repeats a template scrape.

Parameters:

  • name (str): Friendly label for this schedule
  • project_id (str): Project to attach this schedule to
  • schedule (dict): {"type": "cron", "pattern": "0 9 * * *"} or {"type": "interval", "every": <ms>}
  • scrape (dict): Template scrape, e.g. {"url": ..., "schema": ..., "name": ..., "pageType": ...}
  • iterations (int, optional): Cap on how many runs this schedule performs
  • delivery_config (dict, optional): See Delivering Results to a Storage — applies to every future run this schedule generates, not just the template

Returns: Dictionary with scheduleId, templateScrapeId, status, and message

list_schedules()

List all schedules for the authenticated user. Not paginated.

get_schedule(schedule_id)

Retrieve a single schedule by ID.

pause_schedule(schedule_id) / resume_schedule(schedule_id)

Pause or resume a schedule's recurring runs.

delete_schedule(schedule_id)

Stop and permanently delete a schedule.

list_schedule_runs(schedule_id, limit=20, cursor=None)

List the scrapes a schedule has generated so far, cursor-paginated (same shape as list_scrapes).

Utilities

check_status(job_id)

Poll the status of a job.

Parameters:

  • job_id (str): The jobId returned from creation

Returns: Dictionary with completed (bool) and state (str)

wait_for_completion(job_id, timeout=300, poll_interval=1.0)

Wait for a job to complete.

Parameters:

  • job_id (str): The jobId to poll
  • timeout (int): Max seconds to wait
  • poll_interval (float): Seconds between polls

Returns: Status dictionary when completed

Raises: TimeoutError if job doesn't complete

Delivering Results to a Storage

scrape(), scrape_paginated(), fetch(), extract_fetch(), and create_schedule() all accept an optional delivery_config, which uploads the completed run's data to a destination you've already connected via the dashboard's Storages page — the API can't accept raw S3/Google Drive/Dropbox credentials inline, only reference an already-connected one. A destination that isn't connected (or isn't active) is silently dropped, not an error.

from divparser.constants import DESTINATION_S3, DESTINATION_GOOGLE_DRIVE, DESTINATION_DROPBOX

client.scrape(
    url, schema,
    delivery_config={
        "destinations": [DESTINATION_S3, DESTINATION_GOOGLE_DRIVE],
        "format": "csv"  # "json" (default) | "csv" | "xlsx"
    }
)

# create_schedule's delivery_config applies to every future run, not just the template.
client.create_schedule(
    name="Daily check",
    project_id=project_id,
    schedule={"type": "cron", "pattern": "0 9 * * *"},
    scrape={"url": url, "schema": schema},
    delivery_config={"destinations": [DESTINATION_DROPBOX]}
)

fetch()'s content is raw HTML/Markdown, never tabular — use format: "html" or "markdown" instead.

Utility Functions

The divparser.utils module provides helper functions for working with results. Every result has one of three statuses: SUCCESS, FAILED, or REQUIRES_ATTENTION (the long-term parser's self-healing selectors came back under-confident — data still came back, it's just worth a second look). extract_data_from_results() and flatten_results() include both SUCCESS and REQUIRES_ATTENTION rows since both carry real data; only FAILED rows are skipped.

from divparser.utils import (
    extract_data_from_results,
    flatten_results,
    filter_results_by_status,
    get_results_by_url,
    count_requires_attention_results,
    get_result_stats
)

# Flatten nested results (SUCCESS + REQUIRES_ATTENTION rows only)
all_items = flatten_results(results)

# Get statistics
stats = get_result_stats(results)
print(f"Success rate: {stats['success_rate']:.1f}%")
print(f"Usable rate (including needs-review rows): {stats['usable_rate']:.1f}%")
if stats["requires_attention"]:
    print(f"{stats['requires_attention']} result(s) need a second look")

# Group by URL
by_url = get_results_by_url(results)

Saving & Converting Data

The divparser package (and divparser.convert for the lower-level primitives) provide export helpers:

from divparser import save_scrape_as, save_fetch_as, to_json, to_csv, to_xlsx, html_to_markdown

save_scrape_as(results, format, path)

Save a scrape/parse result to disk as "json", "csv", or "xlsx". results is the results list from get_scrape()/get_parse() — rows with usable data (SUCCESS and REQUIRES_ATTENTION) are combined into one flat export via flatten_results(); FAILED rows are skipped.

save_fetch_as(html, format, path)

Save a fetch's raw HTML to disk, either "html" as-is or converted to "markdown".

to_json(data) / to_csv(data) / to_xlsx(data, sheet_name="ScrapedData") / html_to_markdown(html)

Lower-level conversion primitives that return the converted string/bytes instead of writing a file — use these if you want to stream, upload, or otherwise handle the converted data yourself, or are working with a custom dataset instead of a full results list.

Examples

Example 1: Extract Job Listings

from divparser import DivParser

client = DivParser(api_key="your_api_key")

result = client.scrape_and_parse(
    url="https://example-jobs.com/listings",
    schema="""
    Extract the following for each job:
    - job title
    - company name
    - location
    - salary range (if available)
    """,
    name="Job Listings Scrape"
)

for job in result["results"][0]["data"]:
    print(f"{job['title']} at {job['company']} in {job['location']}")

Example 2: Parse Product Information from HTML

html_content = """
<html>
<body>
    <div class="product">
        <h2>Widget Pro</h2>
        <p class="price">$49.99</p>
        <p class="rating">4.5 stars</p>
    </div>
    <div class="product">
        <h2>Widget Lite</h2>
        <p class="price">$19.99</p>
        <p class="rating">4.2 stars</p>
    </div>
</body>
</html>
"""

result = client.parse_and_wait(
    html=html_content,
    schema="Extract product name, price, and rating"
)

for product in result["results"][0]["data"]:
    print(f"{product['name']}: {product['price']} ({product['rating']})")

Example 3: Batch Scraping Multiple Pages

from divparser.utils import flatten_results

pages = [f"https://example.com/products?page={i}" for i in range(1, 4)]

result = client.scrape_paginated(
    urls=pages,
    schema="Extract product ID, name, and price"
)

# Get all products from all pages
all_products = flatten_results(result["results"])
print(f"Total products: {len(all_products)}")

Error Handling

from divparser import DivParser
import requests

client = DivParser(api_key="your_api_key")

try:
    result = client.scrape_and_parse(
        url="https://example.com",
        schema="Extract content"
    )
except requests.exceptions.HTTPError as e:
    print(f"API Error: {e}")
except TimeoutError as e:
    print(f"Job timed out: {e}")

Best Practices

  1. Use Descriptive Schemas: Clear instructions in your schema lead to better extraction
  2. Set Appropriate Timeouts: Complex extractions may need longer timeouts
  3. Batch Operations: Use scrape_paginated for multiple URLs instead of individual requests
  4. Handle Errors: Always catch exceptions for production code
  5. Reuse Clients: Create one client instance and reuse it

API Documentation

For more detailed information, visit DivParser API Reference.

License

MIT

Support

For issues, questions, or feature requests, visit DivParser Support.

Release files for divparser 0.4.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for divparser 0.4.0
File Size Uploaded
divparser-0.4.0.tar.gz 13.5 kB Details

Built distribution (wheel)

Table of built distributions (wheels) for divparser 0.4.0
File Interpreter ABI Platform
divparser-0.4.0-py3-none-any.whl Python 3 none any Details

Total release size: 29.8 kB

Release files / divparser-0.4.0.tar.gz

Download URL divparser-0.4.0.tar.gz
Size 13.5 kB
Tags Source
SHA-256 checksum
How to use checksums
03923aacff644c5d4dfadebae32fcee107014f2190198c3aeca42b7c0fb7d5fd
BLAKE2b-256 checksum
How to use checksums
6041c5d191d341a278a5f6cdbc7663bb8ac3c9ca54292a9f096e6a5fcfe1301b
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via uv/0.12.5 {"installer":{"name":"uv","version":"0.12.5","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":null,"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}

Release files / divparser-0.4.0-py3-none-any.whl

Download URL divparser-0.4.0-py3-none-any.whl
Size 16.3 kB
Tags Python 3
SHA-256 checksum
How to use checksums
288d48925203b52a67bb96a8787e347de5187b2c4e2b4034eb451ab96ebd067e
BLAKE2b-256 checksum
How to use checksums
da6efe525a279a62616f9c45fee3b34de786365ac618c4b4dbf699523db81e52
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via uv/0.12.5 {"installer":{"name":"uv","version":"0.12.5","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":null,"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}

Release history Release notifications | RSS feed

This release

0.4.0 This release

2 release files

0.2.0

2 release files

0.1.0

2 release files

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page