jpmschweitzerandClaude Opus 4.5
a61fbe5a88
Set up Webber - multi-agent AI development system with: - Domain-based project structure (src/domains/, src/shared/) - BaseController pattern with lazy router instantiation - Pydantic Settings configuration with env file support - Logger decorator with temporal benchmarking and trace IDs - UserProvider singleton for request-scoped context - Custom exception hierarchy - Health endpoints (/, /health) Dependencies (CVE checked 2026-01-09): - FastAPI 0.128.0, Starlette 0.50.0, Uvicorn 0.40.0 - Pydantic 2.12.4, PydanticAI 1.40.0 - All packages at latest safe versions Placeholder domains for future implementation: - agents/ (explore, plan, task) - tools/ (file, shell, search) - auth/ (tatlock integration) Port: 8086 (per CONTAINERS.md allocation) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
108 lines
2.9 KiB
Python
108 lines
2.9 KiB
Python
"""
|
|
Custom exception hierarchy.
|
|
|
|
All application exceptions inherit from AppException.
|
|
"""
|
|
from typing import Any, Optional
|
|
|
|
|
|
class AppException(Exception):
|
|
"""Base exception for application errors."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
status_code: int = 500,
|
|
error_code: Optional[str] = None,
|
|
details: Optional[dict[str, Any]] = None,
|
|
):
|
|
self.message = message
|
|
self.status_code = status_code
|
|
self.error_code = error_code or self.__class__.__name__
|
|
self.details = details or {}
|
|
super().__init__(message)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""Convert to JSON-serializable dict."""
|
|
return {
|
|
"error": self.error_code,
|
|
"message": self.message,
|
|
"details": self.details,
|
|
}
|
|
|
|
|
|
class NotFoundError(AppException):
|
|
"""Resource not found."""
|
|
|
|
def __init__(self, resource: str, identifier: Any):
|
|
super().__init__(
|
|
message=f"{resource} not found: {identifier}",
|
|
status_code=404,
|
|
details={"resource": resource, "identifier": str(identifier)},
|
|
)
|
|
|
|
|
|
class ValidationError(AppException):
|
|
"""Input validation failed."""
|
|
|
|
def __init__(self, message: str, field: Optional[str] = None):
|
|
super().__init__(
|
|
message=message,
|
|
status_code=422,
|
|
details={"field": field} if field else {},
|
|
)
|
|
|
|
|
|
class AuthenticationError(AppException):
|
|
"""Authentication required or failed."""
|
|
|
|
def __init__(self, message: str = "Authentication required"):
|
|
super().__init__(message=message, status_code=401)
|
|
|
|
|
|
class AuthorizationError(AppException):
|
|
"""Permission denied."""
|
|
|
|
def __init__(self, message: str = "Permission denied"):
|
|
super().__init__(message=message, status_code=403)
|
|
|
|
|
|
class ConflictError(AppException):
|
|
"""Resource conflict (e.g., duplicate)."""
|
|
|
|
def __init__(self, message: str):
|
|
super().__init__(message=message, status_code=409)
|
|
|
|
|
|
class RateLimitError(AppException):
|
|
"""Rate limit exceeded."""
|
|
|
|
def __init__(self, retry_after: int = 60):
|
|
super().__init__(
|
|
message="Rate limit exceeded",
|
|
status_code=429,
|
|
details={"retry_after": retry_after},
|
|
)
|
|
|
|
|
|
class ServiceUnavailableError(AppException):
|
|
"""External service unavailable."""
|
|
|
|
def __init__(self, service: str, message: Optional[str] = None):
|
|
super().__init__(
|
|
message=message or f"Service unavailable: {service}",
|
|
status_code=503,
|
|
details={"service": service},
|
|
)
|
|
|
|
|
|
class TimeoutError(AppException):
|
|
"""Operation timed out."""
|
|
|
|
def __init__(self, operation: str, timeout_seconds: float):
|
|
super().__init__(
|
|
message=f"Operation timed out: {operation}",
|
|
status_code=504,
|
|
details={"operation": operation, "timeout_seconds": timeout_seconds},
|
|
)
|