"""
askpablos_api.http
HTTP communication and client functionality for the AskPablos API.
This module handles HTTP communication with the AskPablos API, including
request construction, response parsing, error handling, and provides
the main ProxyClient class for API interactions.
"""
import json
from base64 import b64decode
from typing import Optional, Dict, Any
from urllib.parse import urljoin
import requests
from .auth import AuthManager
from .models import ResponseData, RequestOptions
from .validators import ParameterValidator
from .exceptions import APIConnectionError, ResponseError, RequestTimeoutError
from .config import DEFAULT_API_URL
[docs]
class HTTPClient:
"""
Handles low-level HTTP communication with the AskPablos API.
This class manages the HTTP requests, response parsing,
and error handling for the proxy service communication.
"""
[docs]
def __init__(self, auth_manager: AuthManager, api_url: str):
"""
Initialize the HTTP client.
Args:
auth_manager: Authentication manager for request signing
api_url: Base URL for the API service
"""
self.auth_manager = auth_manager
self.api_url = urljoin(api_url.rstrip('/'), '/api/proxy')
[docs]
def send_request(
self,
url: str,
method: str = "GET",
options: Optional[RequestOptions] = None
) -> ResponseData:
"""
Send an HTTP request through the proxy service.
Args:
url: Target URL to fetch through the proxy
method: HTTP method (GET, POST, etc.)
options: Request options and proxy settings
Returns:
ResponseData: Parsed response from the API
Raises:
APIConnectionError: If connection to API fails
ResponseError: If API returns an error response
"""
if options is None:
options = RequestOptions()
# Validate options
options.validate()
# Build the request payload
request_data = self._build_request_payload(
url=url,
method=method,
options=options
)
# Convert to JSON
payload = json.dumps(request_data, separators=(',', ':'), sort_keys=True)
# Build authentication headers
auth_headers = self.auth_manager.build_auth_headers(payload)
try:
# Send the request
response = requests.post(
self.api_url,
data=payload,
headers=auth_headers,
timeout=options.timeout
)
# Handle HTTP errors
if response.status_code != 200:
error_msg = self._extract_error_message(response)
raise ResponseError(response.status_code, error_msg)
# Parse and return the response
return self._parse_response(response, url)
except requests.Timeout as e:
raise RequestTimeoutError("The request to the AskPablos server timed out. "
"Please check your network connection or try increasing the timeout setting.") from e
except requests.RequestException as e:
raise APIConnectionError("Failed to connect to the askpablos server it may be down or unreachable "
"if the problem persists contact the administrator.") from e
@staticmethod
def _build_request_payload(
url: str,
method: str,
options: RequestOptions
) -> Dict[str, Any]:
"""
Build the request payload for the API.
Args:
url: Target URL
method: HTTP method
options: Request options
Returns:
Dict[str, Any]: Complete request payload
"""
payload = {
"url": url,
"method": method.upper(),
"browser": options.browser,
"timeout": options.timeout,
"maxRetries": options.max_retries
}
# Add optional fields if present
optional_fields = ['screenshot', 'operations', 'geoLocation', 'proxyType', 'body', 'headers']
for field in optional_fields:
if field in options.additional_options:
payload[field] = options.additional_options[field]
elif field == 'screenshot' and options.screenshot:
payload[field] = options.screenshot
return payload
def _parse_response(self, response: requests.Response, url: str) -> ResponseData:
"""
Parse the API response into a ResponseData object.
Args:
response: Raw HTTP response from the API
url: The requested URL
Returns:
ResponseData: Parsed response object
"""
api_response = response.json()
decoded_body = b64decode(api_response.get('responseBody')).decode()
return ResponseData(
status_code=response.status_code,
headers=response.headers,
content=decoded_body,
url=url,
elapsed_time=f"{response.elapsed.total_seconds():.2f}s",
encoding=self._extract_encoding(response.headers.get('Content-Type')),
json_data=api_response,
screenshot=api_response.get('screenshots')
)
@staticmethod
def _extract_encoding(content_type: Optional[str]) -> Optional[str]:
if content_type and 'charset=' in content_type:
return content_type.split('charset=')[-1]
return None
@staticmethod
def _extract_error_message(response: requests.Response) -> str:
"""
Extract error message from API response.
Args:
response: HTTP response with error
Returns:
str: Error message
"""
response_body = json.loads(response.content)
try:
if response_body.get('error'):
return response_body.get('error')
else:
return 'No error details provided'
except (ValueError, json.JSONDecodeError):
return f'HTTP {response.status_code} error'
[docs]
class ProxyClient:
"""
High-level client for the AskPablos proxy service.
This class orchestrates the authentication, HTTP communication,
and validation components to provide a clean interface for
making proxy requests.
"""
[docs]
def __init__(self, api_key: str, secret_key: str, api_url: str = DEFAULT_API_URL):
"""
Initialize the proxy client.
Args:
api_key: Your API key from the AskPablos dashboard
secret_key: Your secret key for HMAC signing
api_url: The proxy API base URL
"""
# Initialize components
self.auth_manager = AuthManager(api_key, secret_key)
self.http_client = HTTPClient(self.auth_manager, api_url)
self.validator = ParameterValidator()
[docs]
def request(
self,
url: str,
method: str = "GET",
headers: Optional[Dict[str, str]] = None,
params: Optional[Dict[str, str]] = None,
body: Optional[Any] = None,
options: Optional[Dict[str, Any]] = None,
timeout: int = 30,
max_retries: int = 3
) -> ResponseData:
"""
Send a request through the AskPablos proxy.
Args:
url: Target URL to fetch through the proxy
method: HTTP method (GET or POST)
headers: Custom headers to send to the target URL
params: Query parameters to append to the target URL
body: Request body, only valid when method='POST'
options: Proxy-specific options (browser, screenshot, geoLocation, proxyType, etc.)
timeout: Request timeout in seconds
max_retries: Maximum number of retries on failure
Returns:
ResponseData: Response object with all request results
"""
if options is None:
options = {}
# Validate all parameters and get normalized option values
validated = self.validator.validate_request_params(
url=url,
headers=headers,
browser=options.get("browser", False),
screenshot=options.get("screenshot", False),
timeout=timeout,
method=method,
body=body,
options=options
)
if body is not None:
options["body"] = body
if headers is not None:
options["headers"] = headers
# Apply normalized values (e.g. lowercased geoLocation)
options.update(validated)
request_options = RequestOptions(
browser=options.get("browser", False),
screenshot=options.get("screenshot", False),
timeout=timeout,
max_retries=max_retries,
**{k: v for k, v in options.items() if k not in ["browser", "screenshot"]}
)
# Send the request
return self.http_client.send_request(
url=url,
method=method,
options=request_options
)