""" Simple API Client for OpenResty with Basic Authentication A lightweight Python client for accessing authenticated OpenResty services with JSON response handling and comprehensive error management. """ import logging from typing import Optional, Dict, Any, Tuple, Union import requests from requests import Session, Response from requests.exceptions import ( RequestException, ConnectionError, Timeout, TooManyRedirects, HTTPError, JSONDecodeError ) # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class APIClientError(Exception): """Custom exception for API client errors""" def __init__(self, message: str, status_code: Optional[int] = None): super().__init__(message) self.message = message self.status_code = status_code class NetworkError(APIClientError): """Network-related errors""" pass class AuthenticationError(APIClientError): """Authentication-related errors""" pass class ParseError(APIClientError): """Data parsing errors""" pass class APIClient: """ Simple API client for accessing authenticated OpenResty services. Supports Basic Authentication and JSON response handling with comprehensive error management and session persistence. """ def __init__( self, base_url: str, username: Optional[str] = None, password: Optional[str] = None, timeout: int = 30 ): """ Initialize the API client. Args: base_url: Base URL of the API service username: Username for Basic Authentication password: Password for Basic Authentication timeout: Request timeout in seconds """ self.base_url = base_url.rstrip('/') self.timeout = timeout # Create session for connection pooling self.session = Session() # Configure authentication if username and password: self.session.auth = (username, password) logger.info(f"Basic authentication configured for user: {username}") # Set default headers self.session.headers.update({ 'User-Agent': 'OpenResty-Python-Client/1.0', 'Accept': 'application/json, text/plain, */*', 'Accept-Encoding': 'gzip, deflate' }) logger.info(f"API client initialized for: {base_url}") def get(self, endpoint: str = "/", params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """ Send GET request to specified endpoint. Args: endpoint: API endpoint (default: "/") params: Query parameters Returns: Dictionary containing response data Raises: APIClientError: For API-related errors NetworkError: For network-related errors AuthenticationError: For authentication failures ParseError: For data parsing errors """ url = f"{self.base_url}{endpoint}" try: logger.info(f"GET request to: {url}") response = self.session.get(url, params=params, timeout=self.timeout) # Parse response safely data, error = self._safe_json_parse(response) if error: raise ParseError(f"Failed to parse response: {error}") # Return structured response return { 'status': 'success', 'status_code': response.status_code, 'data': data, 'headers': dict(response.headers), 'url': response.url } except ConnectionError as e: error_msg = f"Connection error: {str(e)}" logger.error(error_msg) raise NetworkError(error_msg) except Timeout as e: error_msg = f"Request timeout: {str(e)}" logger.error(error_msg) raise NetworkError(error_msg) except HTTPError as e: if e.response.status_code == 401: error_msg = "Authentication failed: Invalid credentials" logger.error(error_msg) raise AuthenticationError(error_msg, 401) else: error_msg = f"HTTP error: {str(e)}" logger.error(error_msg) raise APIClientError(error_msg, e.response.status_code) except RequestException as e: error_msg = f"Request failed: {str(e)}" logger.error(error_msg) raise NetworkError(error_msg) def health_check(self) -> Dict[str, Any]: """ Check the health status of the API service. Returns: Dictionary containing health check results Raises: APIClientError: If health check fails """ try: logger.info("Performing health check") response = self.session.get(f"{self.base_url}/health", timeout=self.timeout) if response.status_code == 200: return { 'status': 'healthy', 'status_code': response.status_code, 'message': 'Service is running', 'response_text': response.text.strip() } else: return { 'status': 'unhealthy', 'status_code': response.status_code, 'message': f"Unexpected status code: {response.status_code}" } except Exception as e: logger.error(f"Health check failed: {str(e)}") return { 'status': 'error', 'message': f"Health check failed: {str(e)}" } def _safe_json_parse(self, response: Response) -> Tuple[Optional[Dict[str, Any]], Optional[str]]: """ Safely parse JSON response with comprehensive error handling. Args: response: requests.Response object Returns: Tuple of (parsed_data, error_message) """ try: # Check HTTP status first response.raise_for_status() # Check content type content_type = response.headers.get('content-type', '').lower() if 'application/json' not in content_type: logger.warning(f"Response content-type is not JSON: {content_type}") # Try to parse JSON try: json_data = response.json() logger.debug(f"Successfully parsed JSON response") return json_data, None except JSONDecodeError as e: # For non-JSON responses, return raw text if response.text.strip(): logger.warning(f"Response is not valid JSON, returning raw text") return {'raw_text': response.text}, None else: error_msg = f"Invalid JSON response: {str(e)}" if hasattr(response, 'text'): error_msg += f"\nResponse content (first 200 chars): {response.text[:200]}" return None, error_msg except HTTPError as e: error_msg = f"HTTP error: {str(e)}" return None, error_msg except Exception as e: error_msg = f"Unexpected error parsing response: {str(e)}" return None, error_msg def __enter__(self): """Context manager entry""" return self def __exit__(self, exc_type, exc_val, exc_tb): """Context manager exit - cleanup resources""" self.close() def close(self): """Close the session and cleanup resources""" if self.session: self.session.close() logger.info("API client session closed") def __del__(self): """Destructor - ensure cleanup""" self.close() # Convenience function for quick usage def create_client( base_url: str = "https://airsltd-ocngx.hf.space", username: str = "admin", password: str = "admin123", timeout: int = 30 ) -> APIClient: """ Create a pre-configured API client. Args: base_url: Base URL of the service username: Username for authentication password: Password for authentication timeout: Request timeout Returns: Configured APIClient instance """ return APIClient(base_url, username, password, timeout)