From 9ed1c2e7fbd817fdfa216f452eabd11d6a1a3fc4 Mon Sep 17 00:00:00 2001 From: Adom Hamza Date: Thu, 2 Jul 2026 11:36:58 +0000 Subject: [PATCH 1/2] feature(connectors/bright_data_unlocker): add Bright Data Web Unlocker connector --- bright_data_unlocker/README.md | 92 +++++++ bright_data_unlocker/configuration.json | 8 + bright_data_unlocker/connector.py | 176 +++++++++++++ bright_data_unlocker/helpers/__init__.py | 22 ++ .../helpers/data_processing.py | 141 +++++++++++ bright_data_unlocker/helpers/unlocker.py | 238 ++++++++++++++++++ bright_data_unlocker/helpers/validation.py | 17 ++ 7 files changed, 694 insertions(+) create mode 100644 bright_data_unlocker/README.md create mode 100644 bright_data_unlocker/configuration.json create mode 100644 bright_data_unlocker/connector.py create mode 100644 bright_data_unlocker/helpers/__init__.py create mode 100644 bright_data_unlocker/helpers/data_processing.py create mode 100644 bright_data_unlocker/helpers/unlocker.py create mode 100644 bright_data_unlocker/helpers/validation.py diff --git a/bright_data_unlocker/README.md b/bright_data_unlocker/README.md new file mode 100644 index 0000000..95a554f --- /dev/null +++ b/bright_data_unlocker/README.md @@ -0,0 +1,92 @@ +# Bright Data Web Unlocker Connector Example + +## Connector overview + +This connector syncs web page content from Bright Data's Web Unlocker API to your Fivetran destination. It fetches unlocked page data for one or more URLs, flattens nested JSON responses, and upserts results to an `unlocker_results` table. + +## Requirements + +- [Supported Python versions](https://github.com/fivetran/connector_sdk/blob/main/README.md#requirements) +- Operating system: + - Windows: 10 or later (64-bit only) + - macOS: 13 (Ventura) or later (Apple Silicon [arm64] or Intel [x86_64]) + - Linux: Distributions such as Ubuntu 20.04 or later, Debian 10 or later, or Amazon Linux 2 or later (arm64 or x86_64) + +## Getting started + +Refer to the [Connector SDK Setup Guide](https://fivetran.com/docs/connector-sdk/setup-guide) to get started. + +Note: Ensure that the `configuration.json` file is not checked into version control to protect sensitive information. + +## Features + +- Fetches unlocked web content via Bright Data Web Unlocker API +- Supports multiple URL input formats (single URL, comma-separated, newline-separated, JSON array string) +- Flattens nested JSON structures for analysis +- Dynamically discovers fields from API responses +- Includes retry logic with exponential backoff for transient API errors +- Checkpoints state after each sync + +## Configuration file + +```json +{ + "api_token": "", + "unlocker_url": "", + "country": "us", + "data_format": "markdown", + "zone": "web_unlocker1", + "format_param": "json" +} +``` + +Configuration parameters: + +- `api_token` (required): Your Bright Data API token +- `unlocker_url` (required): URL or URLs to fetch via the Web Unlocker +- `zone` (optional): Bright Data unlocker zone identifier. Defaults to `web_unlocker1` +- `country` (optional): ISO 3166-1 alpha-2 country code. Defaults to `us` +- `method` (optional): HTTP method. Defaults to `GET` +- `format_param` (optional): Response format (`json` or `html`). Defaults to `json` +- `data_format` (optional): Content format such as `markdown` or `html` + +## Requirements file + +This connector does not require any additional Python packages beyond what is pre-installed in the Fivetran environment. + +Note: The `fivetran_connector_sdk:latest` and `requests:latest` packages are pre-installed in the Fivetran environment. To avoid dependency conflicts, do not declare them in your `requirements.txt`. + +## Authentication + +The Bright Data API uses Bearer token authentication. Obtain your API token from the Bright Data dashboard at https://brightdata.com/cp/setting/users. + +## Data handling + +1. Configuration validation — refer to `validate_configuration()` in `helpers/validation.py` +2. URL parsing — refer to `parse_unlocker_urls()` in `connector.py` +3. API requests — refer to `perform_web_unlocker()` in `helpers/unlocker.py` +4. Result processing — refer to `process_unlocker_result()` in `helpers/data_processing.py` +5. Data upsertion — refer to `process_and_upsert_results()` +6. State checkpointing — refer to `op.checkpoint()` in `update()` + +## Error handling + +- Retry logic for transient HTTP errors (408, 429, 500, 502, 503, 504) with exponential backoff — refer to `_execute_unlocker_request()` in `helpers/unlocker.py` +- Non-retryable failures raise `RuntimeError` with API error details +- Primary key validation issues are logged once per sync via `log.warning()` + +## Tables created + +| Table Name | Primary Key | Description | +|---------------------|--------------------------------|----------------------------------------------------------| +| `UNLOCKER_RESULTS` | `requested_url`, `result_index` | Flattened web unlocker results for each requested URL. | + +## Additional files + +- `helpers/validation.py` — Configuration parameter validation +- `helpers/unlocker.py` — Bright Data Web Unlocker API interaction and retry logic +- `helpers/data_processing.py` — Data flattening, field discovery, and upsert utilities + +## Additional considerations + +The examples provided are intended to help you effectively use Fivetran's Connector SDK. While we've tested the code, Fivetran cannot be held responsible for any unexpected or negative consequences that may arise from using these examples. For inquiries, please reach out to our Support team. diff --git a/bright_data_unlocker/configuration.json b/bright_data_unlocker/configuration.json new file mode 100644 index 0000000..74299de --- /dev/null +++ b/bright_data_unlocker/configuration.json @@ -0,0 +1,8 @@ +{ + "api_token": "", + "unlocker_url": "", + "country": "us", + "data_format": "markdown", + "zone": "web_unlocker1", + "format_param": "json" +} diff --git a/bright_data_unlocker/connector.py b/bright_data_unlocker/connector.py new file mode 100644 index 0000000..c8ec944 --- /dev/null +++ b/bright_data_unlocker/connector.py @@ -0,0 +1,176 @@ +"""This connector syncs web unlocker results from Bright Data's Web Unlocker API to Fivetran. +See the Technical Reference documentation +(https://fivetran.com/docs/connectors/connector-sdk/technical-reference#update) +and the Best Practices documentation +(https://fivetran.com/docs/connectors/connector-sdk/best-practices) for details +""" + +# For reading configuration from a JSON file +import json + +# Helper functions for data processing, validation, and API interaction +from helpers import ( + collect_all_fields, + perform_web_unlocker, + process_and_upsert_results, + process_unlocker_result, + validate_configuration, +) + +# For supporting Connector operations like Update() and Schema() +from fivetran_connector_sdk import Connector + +# For enabling Logs in your connector code +from fivetran_connector_sdk import Logging as log + +# For supporting Data operations like Upsert(), Update(), Delete() and checkpoint() +from fivetran_connector_sdk import Operations as op + +__UNLOCKER_TABLE = "unlocker_results" + + +def schema(configuration: dict): + """ + Define the schema function which lets you configure the schema your connector delivers. + See the technical reference documentation for more details on the schema function: + https://fivetran.com/docs/connectors/connector-sdk/technical-reference#schema + Args: + configuration: a dictionary that holds the configuration settings for the connector. + """ + return [ + { + "table": __UNLOCKER_TABLE, + "primary_key": [ + "requested_url", + "result_index", + ], + "columns": { + "requested_url": "STRING", + "result_index": "INT", + }, + } + ] + + +def update(configuration: dict, state: dict): + """ + Define the update function which lets you configure how your connector fetches data. + See the technical reference documentation for more details on the update function: + https://fivetran.com/docs/connectors/connector-sdk/technical-reference#update + Args: + configuration: a dictionary that holds the configuration settings for the connector. + state: a dictionary that holds the state of the connector. + """ + log.warning("Example: Connectors : Bright Data Web Unlocker") + + validate_configuration(configuration=configuration) + + new_state = dict(state) if state else {} + + unlocker_url_input = configuration.get("unlocker_url", "") + urls = parse_unlocker_urls(unlocker_url_input) + + if urls: + sync_unlocker_urls(configuration=configuration, urls=urls, state=new_state) + + # Save the progress by checkpointing the state. This is important for ensuring that the sync + # process can resume from the correct position in case of next sync or interruptions. + # Learn more about how and where to checkpoint by reading our best practices documentation + # (https://fivetran.com/docs/connectors/connector-sdk/best-practices#largedatasetrecommendation). + op.checkpoint(state=new_state) + + +def sync_unlocker_urls(configuration: dict, urls: list, state: dict): + """ + Fetch unlocker results for the requested URLs and upsert them to Fivetran. + Args: + configuration: Configuration dictionary containing unlocker parameters. + urls: List of URLs to unlock/fetch. + state: Current connector state. + """ + api_token = configuration.get("api_token") + country = configuration.get("country") + data_format = configuration.get("data_format") + format_param = configuration.get("format_param") + method = configuration.get("method") or "GET" + unlocker_zone = configuration.get("zone") + + payload = urls if len(urls) > 1 else urls[0] + unlocker_results = perform_web_unlocker( + api_token=api_token, + url=payload, + zone=unlocker_zone, + country=country, + method=method, + format_param=format_param, + data_format=data_format, + ) + + if not isinstance(unlocker_results, list): + unlocker_results = [unlocker_results] + + processed_results = [] + for index, result in enumerate(unlocker_results): + requested_url = result.get("requested_url") if isinstance(result, dict) else None + if not requested_url: + requested_url = urls[index % len(urls)] + processed_results.append(process_unlocker_result(result, requested_url, index)) + + if not processed_results: + log.warning("No unlocker results returned from API") + return + + log.info(f"Upserting {len(processed_results)} unlocker results to Fivetran") + + all_fields = collect_all_fields(processed_results) + process_and_upsert_results(processed_results, all_fields, __UNLOCKER_TABLE) + + state["last_unlocker_urls"] = urls + state["last_unlocker_count"] = len(processed_results) + + +def parse_unlocker_urls(unlocker_url_input) -> list: + """ + Normalize the unlocker_url configuration value into a list of URLs. + Args: + unlocker_url_input: The unlocker_url configuration value (various formats supported). + Returns: + list: List of normalized URL strings. + """ + if not unlocker_url_input: + return [] + + if isinstance(unlocker_url_input, list): + return [ + item.strip() for item in unlocker_url_input if isinstance(item, str) and item.strip() + ] + + if isinstance(unlocker_url_input, str): + try: + parsed = json.loads(unlocker_url_input) + if isinstance(parsed, list): + return [item.strip() for item in parsed if isinstance(item, str) and item.strip()] + if isinstance(parsed, str) and parsed.strip(): + return [parsed.strip()] + except (json.JSONDecodeError, TypeError): + pass + + if "," in unlocker_url_input: + return [item.strip() for item in unlocker_url_input.split(",") if item.strip()] + + if "\n" in unlocker_url_input: + return [item.strip() for item in unlocker_url_input.split("\n") if item.strip()] + + return [unlocker_url_input.strip()] if unlocker_url_input.strip() else [] + + return [] + + +connector = Connector(update=update, schema=schema) + + +if __name__ == "__main__": + with open("configuration.json", "r") as f: + configuration = json.load(f) + + connector.debug(configuration=configuration) diff --git a/bright_data_unlocker/helpers/__init__.py b/bright_data_unlocker/helpers/__init__.py new file mode 100644 index 0000000..3a40972 --- /dev/null +++ b/bright_data_unlocker/helpers/__init__.py @@ -0,0 +1,22 @@ +"""Helper module exports for easy importing.""" + +# For exporting the helper functions +from .data_processing import ( + collect_all_fields, + process_and_upsert_results, + process_unlocker_result, +) + +# For performing the web unlocker +from .unlocker import perform_web_unlocker + +# For validating the configuration +from .validation import validate_configuration + +__all__ = [ + "collect_all_fields", + "perform_web_unlocker", + "process_and_upsert_results", + "process_unlocker_result", + "validate_configuration", +] diff --git a/bright_data_unlocker/helpers/data_processing.py b/bright_data_unlocker/helpers/data_processing.py new file mode 100644 index 0000000..d28ef57 --- /dev/null +++ b/bright_data_unlocker/helpers/data_processing.py @@ -0,0 +1,141 @@ +"""Data processing utilities for flattening and normalizing Bright Data results.""" + +# For parsing JSON payloads +import json + +# For type hints +from typing import Any + +# For enabling Logs and Operations in connector code + +from fivetran_connector_sdk import Logging as log +from fivetran_connector_sdk import Operations as op + + +def flatten_dict( + data: Any, parent_key: str = "", separator: str = "_", max_depth: int = 10 +) -> dict: + """ + Flatten a nested dictionary into a single-level dictionary. + + Converts nested structures like {"a": {"b": 1}} to {"a_b": 1}. + Handles lists by converting them to JSON strings. + Args: + data: The dictionary to flatten. + parent_key: The parent key to use for the flattened dictionary. + separator: The separator to use for the flattened dictionary. + max_depth: The maximum depth to flatten the dictionary to. + Returns: + A dictionary with the flattened dictionary. + """ + if max_depth <= 0: + return {parent_key: json.dumps(data) if data else None} + + items: list = [] + + if isinstance(data, dict): + for key, value in data.items(): + new_key = f"{parent_key}{separator}{key}" if parent_key else key + if isinstance(value, dict): + items.extend(flatten_dict(value, new_key, separator, max_depth - 1).items()) + elif isinstance(value, list): + items.append((new_key, json.dumps(value) if value else "[]")) + else: + items.append((new_key, value)) + elif isinstance(data, list): + return {parent_key: json.dumps(data) if data else "[]"} + else: + return {parent_key: data} + + return dict(items) + + +def collect_all_fields(results: list) -> set: + """Collect all unique field names from a list of result dictionaries. + Args: + results: A list of result dictionaries. + Returns: + A set of all unique field names from the result dictionaries. + """ + all_fields: set = set() + for result in results: + all_fields.update(result.keys()) + return all_fields + + +def process_unlocker_result(result: Any, requested_url: str, result_index: int) -> dict: + """ + Process a single web unlocker result by flattening nested dictionaries. + + Primary key fields (requested_url, result_index) are always preserved and never + overwritten by values from the flattened API response. + Args: + result: The result dictionary to process. + requested_url: The URL that was requested. + result_index: The index of the result. + Returns: + A dictionary with the processed result. + """ + base_fields = { + "requested_url": requested_url, + "result_index": result_index, + "position": result_index + 1, + } + + if not isinstance(result, dict): + base_fields["raw_data"] = str(result) + return base_fields + + flattened = flatten_dict(result) + + for pk_field in ("requested_url", "result_index"): + flattened.pop(pk_field, None) + + final_result = {**flattened, **base_fields} + final_result["result_index"] = int(result_index) + final_result["position"] = int(result_index + 1) + + return final_result + + +def process_and_upsert_results(processed_results: list, all_fields: set, table_name: str) -> None: + """Validate primary keys and upsert processed unlocker result records. + Args: + processed_results: A list of processed unlocker result dictionaries. + all_fields: A set of all field names from the processed results. + table_name: The name of the table to upsert the results into. + """ + primary_keys = {"requested_url": str, "result_index": int} + primary_key_errors = [] + + for result in processed_results: + for pk, pk_type in primary_keys.items(): + if pk not in result: + primary_key_errors.append(f"Primary key '{pk}' missing from result") + result[pk] = pk_type() if pk_type == str else 0 + elif not isinstance(result[pk], pk_type): + try: + if pk_type == str: + result[pk] = str(result[pk]) + elif pk_type == int: + current_value = result[pk] + if isinstance(current_value, str): + cleaned = current_value.strip().strip("[]\"'") + result[pk] = int(cleaned) if cleaned.isdigit() else 0 + else: + result[pk] = int(current_value) + except (ValueError, TypeError): + primary_key_errors.append( + f"Could not convert primary key '{pk}' to {pk_type.__name__}" + ) + result[pk] = pk_type() if pk_type == str else 0 + + row = {field: result.get(field) for field in all_fields} + op.upsert(table=table_name, data=row) + + if primary_key_errors: + unique_errors = list(set(primary_key_errors)) + log.warning( + f"Primary key validation issues: {', '.join(unique_errors[:3])}" + f"{' (and more)' if len(unique_errors) > 3 else ''}" + ) diff --git a/bright_data_unlocker/helpers/unlocker.py b/bright_data_unlocker/helpers/unlocker.py new file mode 100644 index 0000000..091ff0e --- /dev/null +++ b/bright_data_unlocker/helpers/unlocker.py @@ -0,0 +1,238 @@ +"""Bright Data Web Unlocker helper functions.""" + +# For parsing JSON payloads +import json + +# For retry backoff delays +import time + +# For type hints +from typing import Any, Union + +# For making HTTP requests to the Bright Data API +import requests + +# For HTTP error handling +from requests import RequestException, Response + +# For enabling Logs in your connector code +from fivetran_connector_sdk import Logging as log + +__BRIGHT_DATA_BASE_URL = "https://api.brightdata.com" +__DEFAULT_UNLOCKER_ZONE = "web_unlocker1" +__DEFAULT_TIMEOUT_SECONDS = 120 +__RETRY_STATUS_CODES = {408, 429, 500, 502, 503, 504} + + +def _parse_response_payload(response: Response) -> Any: + """Return JSON payload when available, otherwise raw text. + Args: + response: The response from the Bright Data API. + Returns: + A dictionary with the parsed response payload. + """ + try: + return response.json() + except ValueError: + return response.text + + +def _extract_error_detail(response: Response) -> str: + """Extract error detail from a failed Bright Data response. + Args: + response: The response from the Bright Data API. + Returns: + A string with the error detail. + """ + try: + payload = response.json() + if isinstance(payload, dict): + for key in ("error", "message", "detail", "details"): + if key in payload: + return str(payload[key]) + return str(payload) + return str(payload) + except ValueError: + return response.text + + +def perform_web_unlocker( + api_token: str, + url: Union[str, list], + zone: str | None = __DEFAULT_UNLOCKER_ZONE, + country: str | None = "us", + method: str | None = "GET", + format_param: str | None = "json", + data_format: str | None = "markdown", + timeout: int = __DEFAULT_TIMEOUT_SECONDS, + retries: int = 3, + backoff_factor: float = 1.5, +) -> list: + """Invoke Bright Data's Web Unlocker REST API. + Args: + api_token: The Bright Data API token. + url: The URL to unlock. + zone: The zone to use for the unlocker. + country: The country to use for the unlocker. + method: The method to use for the unlocker. + format_param: The format to use for the unlocker. + data_format: The data format to use for the unlocker. + timeout: The timeout to use for the unlocker. + retries: The number of retries to use for the unlocker. + backoff_factor: The backoff factor to use for the unlocker. + Returns: + A list of dictionaries with the unlocked results. + Raises: + ValueError: If the API token is not valid. + TypeError: If the URL is not a string or list of strings. + ValueError: If the URL is empty. + ValueError: If no non-empty URLs are provided. + """ + if not api_token or not isinstance(api_token, str): + raise ValueError("A valid Bright Data API token is required") + + if not url: + raise ValueError("URL cannot be empty") + + if not isinstance(url, (str, list)): + raise TypeError("URL must be a string or list of strings") + + if isinstance(url, list): + urls = [item.strip() for item in url if isinstance(item, str) and item.strip()] + else: + urls = [url.strip()] + + if not urls: + raise ValueError("At least one non-empty URL must be provided") + + zone_identifier = zone or __DEFAULT_UNLOCKER_ZONE + aggregated_results: list = [] + + for single_url in urls: + payload: dict = { + "zone": zone_identifier, + "url": single_url, + "format": format_param or "json", + } + + if country: + payload["country"] = country.lower() + if method: + payload["method"] = method + if data_format: + payload["data_format"] = data_format + + response_payload = _execute_unlocker_request( + api_token=api_token, + payload=payload, + timeout=timeout, + retries=retries, + backoff_factor=backoff_factor, + ) + + aggregated_results.extend(_normalize_unlocker_result(response_payload, single_url)) + + log.info(f"Unlocker completed successfully. Retrieved {len(aggregated_results)} result(s)") + return aggregated_results + + +def _execute_unlocker_request( + api_token: str, + payload: dict, + timeout: int, + retries: int, + backoff_factor: float, +) -> Any: + """Execute a single unlocker API request with retry logic. + Args: + api_token: The Bright Data API token. + payload: The payload to send to the unlocker. + timeout: The timeout to use for the unlocker. + retries: The number of retries to use for the unlocker. + backoff_factor: The backoff factor to use for the unlocker. + Returns: + A dictionary with the unlocked result. + Raises: + RuntimeError: If the Bright Data Unlocker request fails. + """ + headers = { + "Authorization": f"Bearer {api_token}", + "Content-Type": "application/json", + } + + attempt = 0 + backoff = backoff_factor + + while attempt <= retries: + try: + response = requests.post( + f"{__BRIGHT_DATA_BASE_URL}/request?async=true", + headers=headers, + json=payload, + timeout=timeout, + ) + + if response.status_code == 200: + return _parse_response_payload(response) + + if response.status_code in __RETRY_STATUS_CODES and attempt < retries: + attempt += 1 + log.warning( + f"Bright Data Unlocker request retry {attempt}/{retries} " + f"for URL '{payload.get('url')}' (status code: {response.status_code})" + ) + time.sleep(backoff) + backoff *= backoff_factor + continue + + error_detail = _extract_error_detail(response) + raise RuntimeError( + f"Bright Data Unlocker request failed for URL '{payload.get('url')}': " + f"{error_detail}" + ) + + except RequestException as exc: + if attempt < retries: + attempt += 1 + log.warning( + f"Error contacting Bright Data Unlocker API for URL '{payload.get('url')}': " + f"{str(exc)}. Retrying ({attempt}/{retries})" + ) + time.sleep(backoff) + backoff *= backoff_factor + continue + raise RuntimeError( + f"Failed to execute Bright Data Unlocker request for URL " + f"'{payload.get('url')}' after {retries} retries: {str(exc)}" + ) from exc + + raise RuntimeError("Failed to trigger Bright Data Unlocker request after retries") + + +def _normalize_unlocker_result(payload: Any, source_url: str) -> list: + """Normalize the unlocker result. + Args: + payload: The payload to normalize. + source_url: The source URL. + Returns: + A list of dictionaries with the normalized result. + """ + normalized: list = [] + + if isinstance(payload, list): + for item in payload: + normalized.extend(_normalize_unlocker_result(item, source_url)) + return normalized + + if isinstance(payload, dict): + result = payload.copy() + result.setdefault("requested_url", source_url) + return [result] + + if isinstance(payload, str): + try: + parsed = json.loads(payload) + return _normalize_unlocker_result(parsed, source_url) + except ValueError: + pass + return [{"requested_url": source_url, "raw_response": payload}] diff --git a/bright_data_unlocker/helpers/validation.py b/bright_data_unlocker/helpers/validation.py new file mode 100644 index 0000000..d40061e --- /dev/null +++ b/bright_data_unlocker/helpers/validation.py @@ -0,0 +1,17 @@ +"""Configuration validation utilities.""" + + +def validate_configuration(configuration: dict) -> None: + """ + Validate the configuration dictionary to ensure it contains all required parameters. + + Args: + configuration: A dictionary that holds the configuration settings for the connector. + + Raises: + ValueError: If any required configuration parameter is missing. + """ + required_configs = ["api_token", "unlocker_url"] + for key in required_configs: + if key not in configuration or not configuration.get(key): + raise ValueError(f"Missing required configuration value: {key}") From 92ceffc56062442184565e4c7fba04bbdf6b6df3 Mon Sep 17 00:00:00 2001 From: NII HAMZA ADOM <60192416+adomhamza@users.noreply.github.com> Date: Thu, 2 Jul 2026 13:36:57 +0000 Subject: [PATCH 2/2] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- bright_data_unlocker/README.md | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/bright_data_unlocker/README.md b/bright_data_unlocker/README.md index 95a554f..e5e8ea5 100644 --- a/bright_data_unlocker/README.md +++ b/bright_data_unlocker/README.md @@ -77,10 +77,16 @@ The Bright Data API uses Bearer token authentication. Obtain your API token from ## Tables created -| Table Name | Primary Key | Description | -|---------------------|--------------------------------|----------------------------------------------------------| -| `UNLOCKER_RESULTS` | `requested_url`, `result_index` | Flattened web unlocker results for each requested URL. | - +The connector creates a single table named `unlocker_results` with the following schema (refer to the `schema()` function): + + { + "table": "unlocker_results", + "primary_key": ["requested_url", "result_index"], + "columns": { + "requested_url": "STRING", + "result_index": "INT" + } + } ## Additional files - `helpers/validation.py` — Configuration parameter validation