"""Error handling utilities."""
import httpx
from typing import Optional, Any
class OracleApiError(Exception):
"""Oracle API error."""
def __init__(
self,
message: str,
status_code: Optional[int] = None,
error_code: Optional[str] = None,
original_error: Optional[Any] = None,
):
super().__init__(message)
self.status_code = status_code
self.error_code = error_code
self.original_error = original_error
def handle_oracle_api_error(error: Exception) -> None:
"""Handle Oracle API errors and convert to OracleApiError."""
if isinstance(error, httpx.HTTPStatusError):
status = error.response.status_code
try:
data = error.response.json()
except Exception:
data = {}
error_code = data.get("o:errorCode")
if status == 401:
raise OracleApiError(
"Authentication failed. Please verify your Oracle Fusion username and password "
"are correct. If you're using the demo environment, ensure your account has "
"active access.",
status_code=401,
error_code=error_code,
original_error=error,
)
elif status == 403:
raise OracleApiError(
"Access forbidden. Your account does not have permission to access this "
"resource. Please contact your Oracle Fusion administrator.",
status_code=403,
error_code=error_code,
original_error=error,
)
elif status == 404:
raise OracleApiError(
"Resource not found. The requested invoice or endpoint does not exist. "
"Please verify the invoice ID or API endpoint.",
status_code=404,
error_code=error_code,
original_error=error,
)
elif status == 429:
raise OracleApiError(
"Oracle Fusion API rate limit exceeded. Please wait a few moments before "
"retrying.",
status_code=429,
error_code=error_code,
original_error=error,
)
elif status >= 500:
raise OracleApiError(
"Oracle Fusion API is experiencing issues. Please try again later. "
"If the problem persists, contact Oracle support.",
status_code=status,
error_code=error_code,
original_error=error,
)
else:
message = data.get("detail") or data.get("title") or str(error)
raise OracleApiError(
f"Oracle API Error: {message}",
status_code=status,
error_code=error_code,
original_error=error,
)
elif isinstance(error, (httpx.ConnectError, httpx.NetworkError)):
raise OracleApiError(
"Unable to connect to Oracle Fusion API. Please check:\n"
"1. Network connectivity\n"
"2. API endpoint availability\n"
"3. Firewall settings if deployed on restricted network",
error_code="NETWORK_ERROR",
original_error=error,
)
elif isinstance(error, httpx.TimeoutException):
raise OracleApiError(
"Request to Oracle Fusion API timed out. The service may be slow or unavailable. "
"Please try again.",
error_code="TIMEOUT_ERROR",
original_error=error,
)
else:
raise OracleApiError(str(error), original_error=error)
def sanitize_error_for_logging(error: Exception) -> dict[str, Any]:
"""Sanitize error for logging (remove credentials)."""
if isinstance(error, httpx.HTTPStatusError):
return {
"type": type(error).__name__,
"status": error.response.status_code,
"url": str(error.request.url),
"method": error.request.method,
}
elif isinstance(error, Exception):
return {
"type": type(error).__name__,
"message": str(error),
}
return {"error": "Unknown error"}
class ValidationError(Exception):
"""Validation error."""
def __init__(self, message: str, field: str):
super().__init__(message)
self.field = field
