"""Pydantic schemas for chat room management Request and response models for API endpoints """ from pydantic import BaseModel, Field, ConfigDict, field_serializer from typing import Optional, List from datetime import datetime from enum import Enum class IncidentType(str, Enum): """Types of production incidents""" EQUIPMENT_FAILURE = "equipment_failure" MATERIAL_SHORTAGE = "material_shortage" QUALITY_ISSUE = "quality_issue" OTHER = "other" class SeverityLevel(str, Enum): """Incident severity levels""" LOW = "low" MEDIUM = "medium" HIGH = "high" CRITICAL = "critical" class RoomStatus(str, Enum): """Room lifecycle status""" ACTIVE = "active" RESOLVED = "resolved" ARCHIVED = "archived" class MemberRole(str, Enum): """Room member roles""" OWNER = "owner" EDITOR = "editor" VIEWER = "viewer" # Request Schemas class CreateRoomRequest(BaseModel): """Request to create a new incident room""" title: str = Field(..., min_length=1, max_length=255, description="Room title") incident_type: IncidentType = Field(..., description="Type of incident") severity: SeverityLevel = Field(..., description="Severity level") location: Optional[str] = Field(None, max_length=255, description="Incident location") description: Optional[str] = Field(None, description="Detailed description") lots: Optional[List[str]] = Field(None, description="LOT batch numbers") class UpdateRoomRequest(BaseModel): """Request to update room metadata""" title: Optional[str] = Field(None, min_length=1, max_length=255) severity: Optional[SeverityLevel] = None status: Optional[RoomStatus] = None location: Optional[str] = Field(None, max_length=255) description: Optional[str] = None resolution_notes: Optional[str] = None lots: Optional[List[str]] = Field(None, description="LOT batch numbers") class AddMemberRequest(BaseModel): """Request to add a member to a room""" user_id: str = Field(..., description="User email or ID to add") role: MemberRole = Field(..., description="Role to assign") class UpdateMemberRoleRequest(BaseModel): """Request to update a member's role""" role: MemberRole = Field(..., description="New role") class TransferOwnershipRequest(BaseModel): """Request to transfer room ownership""" new_owner_id: str = Field(..., description="User ID of new owner") class RoomFilterParams(BaseModel): """Query parameters for filtering rooms""" status: Optional[RoomStatus] = None incident_type: Optional[IncidentType] = None severity: Optional[SeverityLevel] = None created_after: Optional[datetime] = None created_before: Optional[datetime] = None search: Optional[str] = Field(None, description="Search in title and description") my_rooms: Optional[bool] = Field(False, description="Filter to show only rooms where user is a member") limit: int = Field(20, ge=1, le=100) offset: int = Field(0, ge=0) # Response Schemas class MemberResponse(BaseModel): """Room member information""" user_id: str role: MemberRole added_by: str added_at: datetime removed_at: Optional[datetime] = None model_config = ConfigDict(from_attributes=True) @field_serializer("added_at", "removed_at") def serialize_datetime(self, dt: Optional[datetime]) -> Optional[str]: """Serialize datetime with 'Z' suffix to indicate UTC""" if dt is None: return None return dt.isoformat() + "Z" class RoomResponse(BaseModel): """Complete room information""" room_id: str title: str incident_type: IncidentType severity: SeverityLevel status: RoomStatus location: Optional[str] = None description: Optional[str] = None resolution_notes: Optional[str] = None lots: List[str] = [] created_by: str created_at: datetime resolved_at: Optional[datetime] = None archived_at: Optional[datetime] = None last_activity_at: datetime last_updated_at: datetime ownership_transferred_at: Optional[datetime] = None ownership_transferred_by: Optional[str] = None member_count: int members: Optional[List[MemberResponse]] = None current_user_role: Optional[MemberRole] = None is_member: bool = False is_admin_view: bool = False model_config = ConfigDict(from_attributes=True) @field_serializer( "created_at", "resolved_at", "archived_at", "last_activity_at", "last_updated_at", "ownership_transferred_at" ) def serialize_datetime(self, dt: Optional[datetime]) -> Optional[str]: """Serialize datetime with 'Z' suffix to indicate UTC""" if dt is None: return None return dt.isoformat() + "Z" class RoomListResponse(BaseModel): """Paginated list of rooms""" rooms: List[RoomResponse] total: int limit: int offset: int class AddLotRequest(BaseModel): """Request to add a LOT to a room""" lot: str = Field(..., min_length=1, description="LOT batch number to add") class PermissionResponse(BaseModel): """User permissions in a room""" role: Optional[MemberRole] = None is_admin: bool = False can_read: bool = False can_write: bool = False can_manage_members: bool = False can_transfer_ownership: bool = False can_update_status: bool = False can_delete: bool = False class SuccessResponse(BaseModel): """Generic success response""" message: str