## Security Enhancements (P0) - Add input validation with max_length and numeric range constraints - Implement WebSocket token authentication via first message - Add path traversal prevention in file storage service ## Permission Enhancements (P0) - Add project member management for cross-department access - Implement is_department_manager flag for workload visibility ## Cycle Detection (P0) - Add DFS-based cycle detection for task dependencies - Add formula field circular reference detection - Display user-friendly cycle path visualization ## Concurrency & Reliability (P1) - Implement optimistic locking with version field (409 Conflict on mismatch) - Add trigger retry mechanism with exponential backoff (1s, 2s, 4s) - Implement cascade restore for soft-deleted tasks ## Rate Limiting (P1) - Add tiered rate limits: standard (60/min), sensitive (20/min), heavy (5/min) - Apply rate limits to tasks, reports, attachments, and comments ## Frontend Improvements (P1) - Add responsive sidebar with hamburger menu for mobile - Improve touch-friendly UI with proper tap target sizes - Complete i18n translations for all components ## Backend Reliability (P2) - Configure database connection pool (size=10, overflow=20) - Add Redis fallback mechanism with message queue - Add blocker check before task deletion ## API Enhancements (P3) - Add standardized response wrapper utility - Add /health/ready and /health/live endpoints - Implement project templates with status/field copying ## Tests Added - test_input_validation.py - Schema and path traversal tests - test_concurrency_reliability.py - Optimistic locking and retry tests - test_backend_reliability.py - Connection pool and Redis tests - test_api_enhancements.py - Health check and template tests Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
179 lines
5.0 KiB
Python
179 lines
5.0 KiB
Python
"""Standardized API response wrapper.
|
|
|
|
Provides utility classes and functions for consistent API response formatting
|
|
across all endpoints.
|
|
"""
|
|
from datetime import datetime
|
|
from typing import Any, Generic, Optional, TypeVar
|
|
from pydantic import BaseModel, Field
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class ErrorDetail(BaseModel):
|
|
"""Detailed error information."""
|
|
error_code: str = Field(..., description="Machine-readable error code")
|
|
message: str = Field(..., description="Human-readable error message")
|
|
field: Optional[str] = Field(None, description="Field that caused the error, if applicable")
|
|
details: Optional[dict] = Field(None, description="Additional error details")
|
|
|
|
|
|
class ApiResponse(BaseModel, Generic[T]):
|
|
"""Standard API response wrapper.
|
|
|
|
All API endpoints should return responses in this format for consistency.
|
|
|
|
Attributes:
|
|
success: Whether the request was successful
|
|
data: The actual response data (null for errors)
|
|
message: Human-readable message about the result
|
|
timestamp: ISO 8601 timestamp of the response
|
|
error: Error details if success is False
|
|
"""
|
|
success: bool = Field(..., description="Whether the request was successful")
|
|
data: Optional[T] = Field(None, description="Response data")
|
|
message: Optional[str] = Field(None, description="Human-readable message")
|
|
timestamp: str = Field(
|
|
default_factory=lambda: datetime.utcnow().isoformat() + "Z",
|
|
description="ISO 8601 timestamp"
|
|
)
|
|
error: Optional[ErrorDetail] = Field(None, description="Error details if failed")
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class PaginatedData(BaseModel, Generic[T]):
|
|
"""Paginated data structure."""
|
|
items: list[T] = Field(default_factory=list, description="List of items")
|
|
total: int = Field(..., description="Total number of items")
|
|
page: int = Field(..., description="Current page number (1-indexed)")
|
|
page_size: int = Field(..., description="Number of items per page")
|
|
total_pages: int = Field(..., description="Total number of pages")
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
# Error codes for common scenarios
|
|
class ErrorCode:
|
|
"""Standard error codes for API responses."""
|
|
# Authentication & Authorization
|
|
UNAUTHORIZED = "AUTH_001"
|
|
FORBIDDEN = "AUTH_002"
|
|
TOKEN_EXPIRED = "AUTH_003"
|
|
INVALID_TOKEN = "AUTH_004"
|
|
|
|
# Validation
|
|
VALIDATION_ERROR = "VAL_001"
|
|
INVALID_INPUT = "VAL_002"
|
|
MISSING_FIELD = "VAL_003"
|
|
INVALID_FORMAT = "VAL_004"
|
|
|
|
# Resource
|
|
NOT_FOUND = "RES_001"
|
|
ALREADY_EXISTS = "RES_002"
|
|
CONFLICT = "RES_003"
|
|
DELETED = "RES_004"
|
|
|
|
# Business Logic
|
|
BUSINESS_ERROR = "BIZ_001"
|
|
INVALID_STATE = "BIZ_002"
|
|
LIMIT_EXCEEDED = "BIZ_003"
|
|
DEPENDENCY_ERROR = "BIZ_004"
|
|
|
|
# Server
|
|
INTERNAL_ERROR = "SRV_001"
|
|
DATABASE_ERROR = "SRV_002"
|
|
EXTERNAL_SERVICE_ERROR = "SRV_003"
|
|
RATE_LIMITED = "SRV_004"
|
|
|
|
|
|
def success_response(
|
|
data: Any = None,
|
|
message: Optional[str] = None,
|
|
) -> dict:
|
|
"""Create a successful API response.
|
|
|
|
Args:
|
|
data: The response data
|
|
message: Optional human-readable message
|
|
|
|
Returns:
|
|
Dictionary with standard response structure
|
|
"""
|
|
return {
|
|
"success": True,
|
|
"data": data,
|
|
"message": message,
|
|
"timestamp": datetime.utcnow().isoformat() + "Z",
|
|
"error": None,
|
|
}
|
|
|
|
|
|
def error_response(
|
|
error_code: str,
|
|
message: str,
|
|
field: Optional[str] = None,
|
|
details: Optional[dict] = None,
|
|
) -> dict:
|
|
"""Create an error API response.
|
|
|
|
Args:
|
|
error_code: Machine-readable error code (use ErrorCode constants)
|
|
message: Human-readable error message
|
|
field: Optional field name that caused the error
|
|
details: Optional additional error details
|
|
|
|
Returns:
|
|
Dictionary with standard error response structure
|
|
"""
|
|
return {
|
|
"success": False,
|
|
"data": None,
|
|
"message": message,
|
|
"timestamp": datetime.utcnow().isoformat() + "Z",
|
|
"error": {
|
|
"error_code": error_code,
|
|
"message": message,
|
|
"field": field,
|
|
"details": details,
|
|
},
|
|
}
|
|
|
|
|
|
def paginated_response(
|
|
items: list,
|
|
total: int,
|
|
page: int,
|
|
page_size: int,
|
|
message: Optional[str] = None,
|
|
) -> dict:
|
|
"""Create a paginated API response.
|
|
|
|
Args:
|
|
items: List of items for current page
|
|
total: Total number of items across all pages
|
|
page: Current page number (1-indexed)
|
|
page_size: Number of items per page
|
|
message: Optional human-readable message
|
|
|
|
Returns:
|
|
Dictionary with standard paginated response structure
|
|
"""
|
|
total_pages = (total + page_size - 1) // page_size if page_size > 0 else 0
|
|
|
|
return {
|
|
"success": True,
|
|
"data": {
|
|
"items": items,
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"total_pages": total_pages,
|
|
},
|
|
"message": message,
|
|
"timestamp": datetime.utcnow().isoformat() + "Z",
|
|
"error": None,
|
|
}
|