Source code for askpablos_api.core

"""
askpablos_api.core

Core functionality for the AskPablos proxy API client.

This module provides the main AskPablos class, which serves as a simple interface
for making GET requests through the AskPablos proxy service. The client handles
authentication, error management, and request formatting automatically.
"""

from typing import (
    Any,
    Dict,
    Optional,
    Literal,
    TypeAlias,
)
import logging

from .http import ProxyClient
from .models import ResponseData
from .validators import ParameterValidator
from .exceptions import AskPablosError, ConfigurationError

ProxyType: TypeAlias = Literal['datacenter', 'residential', 'mobile']
HttpMethod: TypeAlias = Literal['GET', 'POST']

logger = logging.getLogger("askpablos_api")


[docs] class AskPablos: """ Simple interface for making GET and POST requests through the AskPablos proxy API. This class provides a clean interface for sending requests through the AskPablos proxy service. It handles authentication, request formatting, and error management automatically. The AskPablos class is designed to be simple and focused - it supports GET requests (the default) and POST requests with an optional body via the ``method`` and ``body`` parameters. Attributes: client (ProxyClient): The underlying client for making API requests. """
[docs] def __init__(self, api_key: str, secret_key: str): """ Initialize the AskPablos API client. Creates a new instance of the AskPablos client with the provided authentication credentials. The client will use these credentials for all subsequent API requests. Args: api_key (str): Your unique API key from the AskPablos dashboard. secret_key (str): Your private secret key used for HMAC signing. Raises: AuthenticationError: If any of the required credentials are missing or invalid. """ self.client = ProxyClient(api_key, secret_key) self.validator = ParameterValidator()
[docs] def get( self, url: str, method: HttpMethod = 'GET', body: Optional[Any] = None, params: Optional[Dict[str, str]] = None, headers: Optional[Dict[str, str]] = None, browser: bool = False, screenshot: bool = False, operations: Optional[list] = None, timeout: int = 30, max_retries: int = 3, geo_location: Optional[str] = None, proxy_type: Optional[ProxyType] = None, **options ) -> ResponseData: """ Send a GET or POST request through the AskPablos proxy. This is the main method for fetching web pages and API endpoints through the AskPablos proxy service. It supports various options for customizing the request behavior, including sending POST requests with a body. Args: url (str): The target URL to fetch. Must be a valid HTTP/HTTPS URL. method (str, optional): HTTP method to use, either 'GET' or 'POST'. Defaults to 'GET'. body (optional): Request body to send when method='POST'. Can be a dict (sent as JSON) or a string. Only valid when method='POST'. params (Dict[str, str], optional): URL query parameters to append. Example: {"page": "1", "limit": "10"} headers (Dict[str, str], optional): Custom headers for the request. browser (bool, optional): Whether to use browser automation for JavaScript rendering. Defaults to False. screenshot (bool, optional): Whether to take a screenshot of the page. Requires browser=True. Defaults to False. operations (list, optional): List of browser operations to perform. Example: [{"task": "waitForElement", "match": {"on": "xpath", "rule": "visible", "value": "//*[@id='element']"}}] timeout (int, optional): Request timeout in seconds. Defaults to 30. max_retries (int, optional): Maximum number of retries on failure. Defaults to 3. geo_location (str, optional): 2-letter ISO country code for geo-targeting (e.g. 'US', 'PK'). proxy_type (str, optional): Proxy type to use. One of 'datacenter', 'residential', 'mobile'. **options: Additional proxy options. Returns: ResponseData: The response object from the API containing: - status_code (int): HTTP status code from the target server - headers (Dict[str, str]): Response headers from target server - content (bytes): Response body/content - url (str): Final URL after any redirects - elapsed_time (str): Time taken to complete the request - encoding (Optional[str]): Response text encoding - json (Optional[Dict[str, Any]]): Parsed JSON data if available - screenshot (Optional[bytes]): Screenshot data if requested Raises: APIConnectionError: If the client cannot connect to the AskPablos API. ResponseError: If the API returns an error status code. AuthenticationError: If authentication fails. ConfigurationError: If there is a configuration issue with the request, such as passing 'body' when method is not 'POST'. """ # Validate browser-specific options self.validator.validate_browser_dependencies( browser=browser, screenshot=screenshot ) # Validate method and body combination self.validator.validate_method(method) self.validator.validate_body(method, body) # Build proxy options proxy_options = { "browser": browser, "screenshot": screenshot, **options } # Add operations if provided if operations is not None: proxy_options["operations"] = operations if geo_location is not None: proxy_options["geoLocation"] = geo_location if proxy_type is not None: proxy_options["proxyType"] = proxy_type try: return self.client.request( url=url, method=method, headers=headers, params=params, body=body, options=proxy_options, timeout=timeout, max_retries=max_retries ) except AskPablosError as e: logger.error(str(e)) raise