"""Pydantic schemas for API request/response validation.""" from pydantic import BaseModel, Field from typing import Optional, Any, List from datetime import datetime # Event Schemas class EventCreate(BaseModel): """Schema for creating a new event.""" session_id: str = Field(..., description="Claude Code session ID") event_type: str = Field( ..., description="Event type: PreToolUse, PostToolUse, SessionStart, SessionEnd, SubagentStop, UserPromptSubmit", ) tool_name: Optional[str] = Field(None, description="Tool name (Bash, Read, Write, etc.)") tool_input: Optional[Any] = Field(None, description="Tool input parameters (JSON)") tool_output: Optional[str] = Field(None, description="Tool output/response") success: Optional[bool] = Field(None, description="Success/failure status") description: Optional[str] = Field(None, description="Human-readable description") timestamp: float = Field(..., description="Unix timestamp with milliseconds") class Config: json_schema_extra = { "example": { "session_id": "abc-123-def", "event_type": "PostToolUse", "tool_name": "Bash", "tool_input": {"command": "ls -la"}, "tool_output": "total 24\ndrwxr-xr-x...", "success": True, "timestamp": 1734251599.887, } } class EventResponse(BaseModel): """Schema for event response.""" id: int session_id: str event_type: str tool_name: Optional[str] = None tool_input: Optional[Any] = None tool_output: Optional[str] = None success: Optional[bool] = None description: Optional[str] = None timestamp: float created_at: datetime class Config: from_attributes = True class EventListResponse(BaseModel): """Schema for paginated event list response.""" events: List[EventResponse] total: int page: int page_size: int has_more: bool # Statistics Schemas class StatisticsResponse(BaseModel): """Schema for statistics response.""" total_events: int = Field(..., description="Total number of events") total_tools_used: int = Field(..., description="Number of unique tools used") total_agents: int = Field(..., description="Number of agents spawned") total_sessions: int = Field(..., description="Number of unique sessions") last_updated: float = Field(..., description="Last update timestamp") class Config: from_attributes = True json_schema_extra = { "example": { "total_events": 66, "total_tools_used": 3, "total_agents": 3, "total_sessions": 1, "last_updated": 1734251599.887, } } class ToolUsageResponse(BaseModel): """Schema for tool usage statistics.""" tool_name: str usage_count: int first_used: float last_used: float class Config: from_attributes = True class SessionResponse(BaseModel): """Schema for session response.""" session_id: str started_at: float ended_at: Optional[float] = None event_count: int created_at: datetime class Config: from_attributes = True # WebSocket Message Schemas class WebSocketMessage(BaseModel): """Schema for WebSocket messages.""" type: str = Field(..., description="Message type: event_created, stats_updated, connection_status, stats_reset") event: Optional[EventResponse] = Field(None, description="Event data (for event_created)") statistics: Optional[StatisticsResponse] = Field(None, description="Statistics data") message: Optional[str] = Field(None, description="Optional message") class Config: json_schema_extra = { "example": { "type": "event_created", "event": { "id": 123, "session_id": "abc-123", "event_type": "PostToolUse", "tool_name": "Bash", "timestamp": 1734251599.887, }, "statistics": { "total_events": 67, "total_tools_used": 3, "total_agents": 3, "total_sessions": 1, }, } } # Health Check Schema class HealthResponse(BaseModel): """Schema for health check response.""" status: str = Field(..., description="Service status") timestamp: float = Field(..., description="Current timestamp") class Config: json_schema_extra = {"example": {"status": "healthy", "timestamp": 1734251599.887}}