"""
HTTP client for making API calls to OpenAPI endpoints.
"""
import json
import logging
from typing import Any, Dict, Optional, Union
from urllib.parse import urljoin
import httpx
from .models import APIOperation, HTTPMethod
from .config import AuthConfig
# Set up logger for detailed API call logging
logger = logging.getLogger(__name__)
class APIClient:
"""HTTP client for OpenAPI operations."""
def __init__(
self,
base_url: str,
auth: Optional[AuthConfig] = None,
timeout: int = 30,
headers: Optional[Dict[str, str]] = None
):
self.base_url = base_url.rstrip("/")
self.auth = auth
self.timeout = timeout
self.default_headers = headers or {}
# Initialize httpx async client
self._client = httpx.AsyncClient(timeout=timeout)
async def call_operation(
self,
operation: APIOperation,
path_params: Optional[Dict[str, Any]] = None,
query_params: Optional[Dict[str, Any]] = None,
headers: Optional[Dict[str, str]] = None,
body: Optional[Union[Dict[str, Any], str, bytes]] = None
) -> Dict[str, Any]:
"""Make an HTTP call for the given operation."""
# Build URL
url = self._build_url(operation.path, path_params or {})
# Prepare headers
request_headers = self._prepare_headers(headers)
# Prepare request body
request_body = self._prepare_body(operation, body)
# Log the request details
logger.info(f"🔧 Calling API: {operation.operation_id}")
logger.info(f" Method: {operation.method.value.upper()}")
logger.info(f" URL: {url}")
if path_params:
logger.info(f" Path Params: {json.dumps(path_params, indent=2)}")
if query_params:
logger.info(f" Query Params: {json.dumps(query_params, indent=2)}")
if request_body:
if isinstance(request_body, str):
try:
# Try to parse and pretty print JSON
body_obj = json.loads(request_body)
logger.info(f" Request Body: {json.dumps(body_obj, indent=2)}")
except json.JSONDecodeError:
logger.info(f" Request Body: {request_body}")
else:
logger.info(f" Request Body: {request_body}")
# Make the request
try:
response_data = await self._make_request(
method=operation.method,
url=url,
headers=request_headers,
params=query_params,
body=request_body
)
# Log the response
logger.info(f"✅ API Response for {operation.operation_id}:")
logger.info(f" {json.dumps(response_data, indent=2)}")
return {
"status": "success",
"data": response_data,
"operation": operation.operation_id
}
except Exception as e:
logger.error(f"❌ API Error for {operation.operation_id}: {str(e)}")
return {
"status": "error",
"error": str(e),
"operation": operation.operation_id
}
def _build_url(self, path: str, path_params: Dict[str, Any]) -> str:
"""Build the full URL with path parameters."""
# Replace path parameters
url_path = path
for param_name, param_value in path_params.items():
placeholder = f"{{{param_name}}}"
url_path = url_path.replace(placeholder, str(param_value))
return urljoin(self.base_url + "/", url_path.lstrip("/"))
def _prepare_headers(self, additional_headers: Optional[Dict[str, str]] = None) -> Dict[str, str]:
"""Prepare request headers with authentication and defaults."""
headers = self.default_headers.copy()
if additional_headers:
headers.update(additional_headers)
# Add authentication headers
if self.auth:
auth_headers = self._get_auth_headers()
headers.update(auth_headers)
return headers
def _get_auth_headers(self) -> Dict[str, str]:
"""Get authentication headers based on auth config."""
if not self.auth:
return {}
headers = {}
if self.auth.type == "bearer" and self.auth.token:
headers["Authorization"] = f"Bearer {self.auth.token}"
elif self.auth.type == "basic" and self.auth.username and self.auth.password:
import base64
credentials = f"{self.auth.username}:{self.auth.password}"
encoded = base64.b64encode(credentials.encode()).decode()
headers["Authorization"] = f"Basic {encoded}"
elif self.auth.type == "apikey":
if self.auth.header_name and self.auth.token:
headers[self.auth.header_name] = self.auth.token
return headers
def _prepare_body(
self,
operation: APIOperation,
body: Optional[Union[Dict[str, Any], str, bytes]]
) -> Optional[Union[str, bytes]]:
"""Prepare the request body."""
if body is None:
return None
# If it's already a string or bytes, return as-is
if isinstance(body, (str, bytes)):
return body
# For JSON content type, serialize to JSON
if operation.request_body and operation.request_body.content_type == "application/json":
return json.dumps(body)
# For form data, we'd need to handle differently
# For now, just return JSON string
return json.dumps(body)
async def _make_request(
self,
method: HTTPMethod,
url: str,
headers: Dict[str, str],
params: Optional[Dict[str, Any]] = None,
body: Optional[Union[str, bytes]] = None
) -> Any:
"""Make the actual HTTP request."""
# Prepare the request
method_str = method.value.upper()
# Add content type for POST/PUT/PATCH requests with body
if body and method_str in ["POST", "PUT", "PATCH"]:
if "content-type" not in {k.lower(): v for k, v in headers.items()}:
headers["Content-Type"] = "application/json"
try:
# Make the actual HTTP request
response = await self._client.request(
method=method_str,
url=url,
headers=headers,
params=params,
content=body
)
# Log response status
logger.info(f" Response Status: {response.status_code}")
# Handle response
if response.status_code >= 400:
error_text = response.text
logger.error(f" Error Response: {error_text}")
raise httpx.HTTPStatusError(
message=f"HTTP {response.status_code}: {error_text}",
request=response.request,
response=response
)
# Try to parse JSON response
try:
response_data = response.json()
except ValueError:
# If not JSON, return the text
response_data = response.text
return response_data
except httpx.RequestError as e:
logger.error(f" Request Error: {str(e)}")
raise httpx.RequestError(f"Request failed: {str(e)}")
except httpx.HTTPStatusError as e:
logger.error(f" HTTP Error: {str(e)}")
raise httpx.HTTPStatusError(
message=f"HTTP error: {str(e)}",
request=e.request,
response=e.response
)
async def close(self):
"""Close the HTTP client."""
if self._client:
await self._client.aclose()
