feature(connectors/bright_data_scrape): add Bright Data Web Scraper C…#41
feature(connectors/bright_data_scrape): add Bright Data Web Scraper C…#41adomhamza wants to merge 9 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR adds a new Connector SDK example (bright_data_scrape) that syncs web scraping results from Bright Data’s Web Scraper API into a Fivetran destination, including helper modules for scraping/polling, validation, and result flattening.
Changes:
- Introduces a new Bright Data scraper connector (
connector.py) with schema, URL parsing/validation, state checkpointing, and upsert logic. - Adds helper modules for API job triggering + snapshot polling (
helpers/scrape.py), result flattening/field discovery (helpers/data_processing.py), and configuration validation (helpers/validation.py). - Adds end-user documentation (
README.md) and a placeholderconfiguration.json.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 18 comments.
Show a summary per file
| File | Description |
|---|---|
| bright_data_scrape/connector.py | Core Connector SDK implementation (schema/update, URL parsing, sync/upsert, state). |
| bright_data_scrape/helpers/scrape.py | Bright Data API trigger + snapshot polling + response parsing utilities. |
| bright_data_scrape/helpers/data_processing.py | Flattening and row-shaping utilities for scraped results. |
| bright_data_scrape/helpers/validation.py | Configuration validation helper. |
| bright_data_scrape/helpers/init.py | Helper re-exports for simplified imports. |
| bright_data_scrape/configuration.json | Placeholder configuration values (no secrets). |
| bright_data_scrape/README.md | Example documentation (setup/configuration/usage). |
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
|
Hi @adomhamza , Could you please rebase your PR so that the latest checks can run and pass Thank you! |
Done. |
Co-authored-by: Jenas Anton Vimal <jenas.vimal@fivetran.com>
Co-authored-by: Jenas Anton Vimal <jenas.vimal@fivetran.com>
Co-authored-by: Jenas Anton Vimal <jenas.vimal@fivetran.com>
… on version control for configuration files.
…_scrape' into feature/bright_data_scrape
… rearranging imports and enhancing list comprehensions. Added logging for URL parsing errors and streamlined the process of handling scrape results.
fivetran-anushkaparashar
left a comment
fivetran-anushkaparashar
left a comment
There was a problem hiding this comment.
I've added a few comment, please have a look!
| return results | ||
|
|
||
|
|
||
| def _parse_json_response( |
There was a problem hiding this comment.
Response body consumed before fallback parse. In _parse_snapshot_response, when Content-Length exceeds the threshold, _parse_large_json_array_streaming reads the body via response.iter_content(). If it returns None (body is not a JSON array), control falls through to _parse_json_response(response, ...) which calls response.json() on an already-drained stream — this always fails or returns garbage for responses over 100 MB that aren't JSON arrays.
If the chunked path is attempted but returns None, do not call _parse_json_response. Either buffer the full content during chunked reading and pass the string directly, or return None/raise before calling the fallback.
| log.info(f"Snapshot {snapshot_id[:8]}... still processing (attempt {attempt})") | ||
|
|
||
|
|
||
| def _handle_non_200_response(response: Response, snapshot_id: str, attempt: int) -> None: |
There was a problem hiding this comment.
All non-200, non-404 poll errors are silently retried. _handle_non_200_response calls response.raise_for_status(), which throws requests.exceptions.HTTPError — a subclass of RequestException. The except RequestException handler in _poll_snapshot catches it, logs every 5th attempt, sleeps, and continues. A 401 Unauthorized or 403 Forbidden will therefore be retried up to max_attempts (100) times (~50 minutes at 30s interval) before hitting the timeout error, hiding the actual failure.
In _handle_non_200_response (or inline in the polling loop), re-raise non-transient HTTP status codes (e.g., 401, 403, 400) immediately without letting them reach the except RequestException catch.
| if max_depth <= 0: | ||
| return {parent_key: json.dumps(data) if data else None} |
| missing_results = [] | ||
| for url_idx, url in enumerate(urls): | ||
| if url_idx < len(scrape_results): | ||
| result = scrape_results[url_idx] |
There was a problem hiding this comment.
The Bright Data batch API returns a flat list; if results are reordered, fewer results than URLs are returned, or a URL produces zero items, data is silently misattributed (wrong URL attached to a result) or silently dropped.
Either rely on the URL embedded in the result payload (result.get("input", {}).get("url") or result.get("url")) to attribute results, or add a post-match assertion when the result count diverges from the URL count beyond a warning.
| process_scrape_result(result, result_url, result_idx) | ||
| ) | ||
| elif isinstance(result, list): | ||
| for item_idx, item in enumerate(result): |
There was a problem hiding this comment.
Duplicate primary key collision when a single URL returns nested-list results. In the single-URL branch, when a top-level result is a list, it enumerates items with item_idx starting from 0. If multiple top-level results are themselves lists, item_idx resets to 0 for each, producing duplicate (url, result_index) pairs — the later upsert silently overwrites earlier rows.
Maintain a monotonically increasing global counter across all nested results instead of resetting item_idx per sub-list.
| required_configs = ["api_token", "dataset_id", "scrape_url"] | ||
| for key in required_configs: | ||
| if key not in configuration: | ||
| raise ValueError(f"Missing required configuration value: {key}") |
| snapshot_id: Snapshot ID to poll | ||
| poll_interval: Interval between polling attempts in seconds | ||
| timeout: Request timeout in seconds | ||
| max_attempts: Maximum number of polling attempts before raising an error (default: 1000) |
| process_and_upsert_results(processed_results, all_fields) | ||
|
|
||
| # Update state with sync information | ||
| state["last_scrape_urls"] = valid_urls |
There was a problem hiding this comment.
state["last_scrape_urls"] and state["last_scrape_count"] are checkpointed each sync, but update() never reads them.
Implement actual incremental logic using states.
…onnector.
Description of Change
/datasets/v3/trigger, polls snapshot completion, flattens nested JSON responses, and upserts results to ascrape_resultstable with primary key(url, result_index).schema(), scalarop.upsert()calls, retry logic with exponential backoff, configuration validation, URL validation, state checkpointing, and key-presence checks during snapshot/JSONL data extraction. No extra third-party dependencies beyond the pre-installed Fivetran environment.Features
log.warning()before syncFiles added
connector.py—schema(),update(), URL parsing/validation, result processing, upsert logic, debug entry pointhelpers/scrape.py— job triggering, snapshot polling, retry logic, response parsinghelpers/data_processing.py— flattening and field discovery utilitieshelpers/validation.py— configuration validationconfiguration.json— placeholder values (no secrets)README.md— setup, configuration, and usage documentationTesting
Checklist
Some tips and links to help validate your PR:
fivetran debugcommand.