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 URLschema(str): Extraction instructions (plain English or Nestlang)name(str, optional): Friendly label for this scrapepage_type(str): "LISTING" (default) or "DETAIL"wait(bool): Wait for completion before returningtimeout(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 scrapeschema(str): Extraction instructionsname(str, optional): Friendly labelpage_type(str): "LISTING" or "DETAIL"wait(bool): Wait for completiontimeout(int): Max seconds to waitdelivery_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 parseschema(str): Extraction instructionsname(str, optional): Friendly labelwait(bool): Wait for completiontimeout(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 URLname(str, optional): Friendly labelproject_id(str, optional): Project to attach this fetch toproxy_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 fromfetch()
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 fromfetch()schema(str): Extraction instructionsdelivery_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 scheduleproject_id(str): Project to attach this schedule toschedule(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 performsdelivery_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 polltimeout(int): Max seconds to waitpoll_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
- Use Descriptive Schemas: Clear instructions in your schema lead to better extraction
- Set Appropriate Timeouts: Complex extractions may need longer timeouts
- Batch Operations: Use
scrape_paginatedfor multiple URLs instead of individual requests - Handle Errors: Always catch exceptions for production code
- 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)
| File | Size | Uploaded | |
|---|---|---|---|
| divparser-0.4.0.tar.gz | 13.5 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| 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}
|