vanitha
(staff)-Added convert_objectid_to_str utility for BSON serialization
544529e
"""
Response formatting utilities for standardized API responses.
"""
from typing import Any, Dict, Optional, List
from datetime import datetime
from uuid import uuid4
def success_response(
data: Any,
message: str = "Success",
correlation_id: Optional[str] = None
) -> Dict[str, Any]:
"""
Format a successful API response.
Args:
data: Response data
message: Success message
correlation_id: Optional correlation ID for request tracing
Returns:
Standardized success response dictionary
"""
return {
"success": True,
"data": data,
"message": message,
"correlation_id": correlation_id or str(uuid4()),
"timestamp": datetime.utcnow().isoformat() + "Z"
}
def error_response(
error: str,
detail: str,
errors: Optional[List[Dict[str, Any]]] = None,
code: Optional[str] = None, # Kept for backward compatibility if needed, but mapped to error title usually
request_id: Optional[str] = None,
headers: Optional[Dict[str, str]] = None
) -> Dict[str, Any]:
"""
Format an error API response according to the Error Handling Guide.
Args:
error: Error title (e.g., "Validation Error", "Authentication Error")
detail: Human-readable error message
errors: Optional list of specific errors (e.g., validation fields)
code: Optional error code (legacy support, can be ignored or used as error title)
request_id: Optional request ID
headers: Optional headers (included in body as per guide example)
Returns:
Standardized error response dictionary
"""
response = {
"success": False,
"error": error or code or "Error",
"detail": detail
}
if errors:
response["errors"] = errors
if request_id:
response["request_id"] = request_id
if headers:
response["headers"] = headers
response["timestamp"] = datetime.utcnow().isoformat() + "Z"
return response
def paginated_response(
documents: list,
total: int,
page: int,
limit: int,
correlation_id: Optional[str] = None
) -> Dict[str, Any]:
"""
Format a paginated list response.
Args:
documents: List of documents for current page
total: Total number of documents
page: Current page number
limit: Page size limit
correlation_id: Optional correlation ID
Returns:
Standardized paginated response
"""
has_more = (page * limit) < total
return success_response(
data={
"documents": documents,
"total": total,
"has_more": has_more,
"page": page,
"limit": limit
},
correlation_id=correlation_id
)