"""42crunch API client."""
import httpx
from typing import Optional, Dict, Any
from .config import Config
class FortyTwoCrunchClient:
"""Client for interacting with the 42crunch API."""
def __init__(self, config: Optional[Config] = None):
"""Initialize the API client.
Args:
config: Configuration instance. If None, creates a new one.
"""
self.config = config or Config()
self.client = httpx.Client(
base_url=self.config.BASE_URL,
headers=self.config.get_auth_headers(),
timeout=30.0,
)
def _handle_response(self, response: httpx.Response) -> Dict[str, Any]:
"""Handle API response and raise appropriate errors.
Args:
response: HTTP response object
Returns:
JSON response data
Raises:
httpx.HTTPStatusError: For HTTP errors
"""
# Provide more detailed error information
if not response.is_success:
error_detail = f"HTTP {response.status_code}"
try:
error_body = response.json()
if isinstance(error_body, dict):
error_msg = error_body.get("message") or error_body.get("error") or str(error_body)
error_detail = f"{error_detail}: {error_msg}"
else:
error_detail = f"{error_detail}: {error_body}"
except Exception:
error_detail = f"{error_detail}: {response.text[:200]}"
# Raise with detailed error message
response.raise_for_status()
return response.json()
def list_collections(
self,
page: int = 1,
per_page: int = 10,
order: str = "default",
sort: str = "default",
) -> Dict[str, Any]:
"""List all API collections.
Args:
page: Page number (default: 1)
per_page: Items per page (default: 10)
order: Sort order (default: "default")
sort: Sort field (default: "default")
Returns:
Dictionary containing collections data
"""
params = {
"page": page,
"perPage": per_page,
"order": order,
"sort": sort,
}
response = self.client.get(
f"{self.config.API_V1_BASE}/search/collections",
params=params,
)
return self._handle_response(response)
def get_collection_apis(
self,
collection_id: str,
with_tags: bool = True,
per_page: int = 0,
) -> Dict[str, Any]:
"""Get all APIs within a collection.
Args:
collection_id: Collection UUID
with_tags: Include tags in response (default: True)
per_page: Items per page (0 = all, default: 0)
Returns:
Dictionary containing APIs data
"""
params = {
"withTags": str(with_tags).lower(),
"perPage": per_page,
}
response = self.client.get(
f"{self.config.API_V2_BASE}/collections/{collection_id}/apis",
params=params,
)
return self._handle_response(response)
def get_api_details(
self,
api_id: str,
branch: str = "main",
read_tags: bool = True,
read_openapi_definition: bool = True,
read_assessment: bool = True,
read_scan: bool = True,
) -> Dict[str, Any]:
"""Get detailed information about a specific API.
Args:
api_id: API UUID
branch: Branch name (default: "main")
read_tags: Include tags (default: True)
read_openapi_definition: Include OpenAPI definition (default: True)
read_assessment: Include assessment data (default: True)
read_scan: Include scan results (default: True)
Returns:
Dictionary containing complete API details
"""
params = {
"readTags": str(read_tags).lower(),
"readOpenApiDefinition": str(read_openapi_definition).lower(),
"readAssessment": str(read_assessment).lower(),
"readScan": str(read_scan).lower(),
}
response = self.client.get(
f"{self.config.API_V2_BASE}/apis/{api_id}/branches/{branch}",
params=params,
)
return self._handle_response(response)
def close(self) -> None:
"""Close the HTTP client."""
self.client.close()
def __enter__(self):
"""Context manager entry."""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit."""
self.close()
