"""HTTP client for OpenProject API v3."""
import base64
from typing import Any
import httpx
from .config import OpenProjectConfig
from .utils.errors import (
OpenProjectAuthError,
OpenProjectConflictError,
OpenProjectError,
OpenProjectNotFoundError,
OpenProjectPermissionError,
OpenProjectRateLimitError,
OpenProjectValidationError,
)
class OpenProjectClient:
"""HTTP client for OpenProject API v3.
Handles authentication, request formatting, and error handling for
OpenProject API calls.
"""
def __init__(self, config: OpenProjectConfig | None = None):
"""Initialize the OpenProject client.
Args:
config: Configuration object, will be loaded from env if not provided
"""
self.config = config or OpenProjectConfig()
self.base_url = self.config.openproject_url.rstrip("/")
self.api_base = f"{self.base_url}/api/v3"
# Create Basic Auth header: base64(apikey:api_key_value)
credentials = f"apikey:{self.config.openproject_api_key}"
auth_bytes = credentials.encode("ascii")
auth_b64 = base64.b64encode(auth_bytes).decode("ascii")
self.headers = {
"Authorization": f"Basic {auth_b64}",
"Content-Type": "application/json",
"Accept": "application/hal+json",
}
self.client = httpx.AsyncClient(
headers=self.headers,
timeout=self.config.openproject_timeout,
follow_redirects=True,
)
async def _handle_response(self, response: httpx.Response) -> dict[str, Any]:
"""Handle API response and raise appropriate exceptions.
Args:
response: HTTP response from OpenProject API
Returns:
Parsed JSON response as dict
Raises:
OpenProjectAuthError: Authentication failed (401)
OpenProjectPermissionError: Insufficient permissions (403)
OpenProjectNotFoundError: Resource not found (404)
OpenProjectConflictError: Conflict, usually lock_version mismatch (409)
OpenProjectValidationError: Invalid request data (422)
OpenProjectRateLimitError: Rate limit exceeded (429)
OpenProjectError: Other API errors
"""
if response.status_code in (200, 201):
return response.json()
elif response.status_code == 204:
# No content - successful deletion
return {"success": True}
elif response.status_code == 401:
raise OpenProjectAuthError("Invalid API key or authentication failed")
elif response.status_code == 403:
raise OpenProjectPermissionError(
f"Insufficient permissions: {response.text}"
)
elif response.status_code == 404:
raise OpenProjectNotFoundError(f"Resource not found: {response.url}")
elif response.status_code == 409:
raise OpenProjectConflictError(
"Conflict detected. The resource may have been modified. "
"Please fetch the latest version and retry with the correct lock_version."
)
elif response.status_code == 422:
try:
error_data = response.json()
errors = error_data.get("_embedded", {}).get("errors", [])
if errors:
error_messages = [
f"{err.get('message', 'Unknown error')}" for err in errors
]
raise OpenProjectValidationError(
f"Validation failed: {'; '.join(error_messages)}"
)
raise OpenProjectValidationError(f"Validation failed: {response.text}")
except ValueError:
raise OpenProjectValidationError(f"Validation failed: {response.text}")
elif response.status_code == 429:
raise OpenProjectRateLimitError(
"Rate limit exceeded. Please wait before retrying."
)
else:
raise OpenProjectError(
f"Unexpected error ({response.status_code}): {response.text}"
)
async def get(
self, endpoint: str, params: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Send GET request to OpenProject API.
Args:
endpoint: API endpoint path (e.g., 'work_packages/123')
params: Optional query parameters
Returns:
Parsed JSON response
"""
url = f"{self.api_base}/{endpoint.lstrip('/')}"
response = await self.client.get(url, params=params)
return await self._handle_response(response)
async def post(self, endpoint: str, data: dict[str, Any]) -> dict[str, Any]:
"""Send POST request to OpenProject API.
Args:
endpoint: API endpoint path
data: Request body data
Returns:
Parsed JSON response
"""
url = f"{self.api_base}/{endpoint.lstrip('/')}"
response = await self.client.post(url, json=data)
return await self._handle_response(response)
async def patch(self, endpoint: str, data: dict[str, Any]) -> dict[str, Any]:
"""Send PATCH request to OpenProject API.
Args:
endpoint: API endpoint path
data: Request body data with fields to update
Returns:
Parsed JSON response
"""
url = f"{self.api_base}/{endpoint.lstrip('/')}"
response = await self.client.patch(url, json=data)
return await self._handle_response(response)
async def delete(self, endpoint: str) -> dict[str, Any]:
"""Send DELETE request to OpenProject API.
Args:
endpoint: API endpoint path
Returns:
Success confirmation dict
"""
url = f"{self.api_base}/{endpoint.lstrip('/')}"
response = await self.client.delete(url)
return await self._handle_response(response)
async def close(self):
"""Close the HTTP client connection."""
await self.client.aclose()
async def __aenter__(self):
"""Async context manager entry."""
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Async context manager exit."""
await self.close()
