from datetime import datetime from typing import Optional, Any, Literal from uuid import UUID from pydantic import BaseModel, Field, field_validator, model_validator import re # --- Tree Type --- TreeType = Literal['troubleshooting', 'procedural', 'maintenance'] # --- Intake Form Schemas --- FIELD_TYPES = Literal[ 'text', 'textarea', 'number', 'ip_address', 'email', 'url', 'select', 'multi_select', 'checkbox', 'password' ] VARIABLE_NAME_PATTERN = re.compile(r'^[a-z][a-z0-9_]*$') class IntakeFieldValidation(BaseModel): """Validation config for an intake form field.""" min_length: Optional[int] = None max_length: Optional[int] = None pattern: Optional[str] = None pattern_message: Optional[str] = None format: Optional[Literal['ipv4', 'email']] = None min_value: Optional[float] = None max_value: Optional[float] = None min_selections: Optional[int] = None max_selections: Optional[int] = None class IntakeFormField(BaseModel): """Schema for a single intake form field definition.""" variable_name: str = Field(..., min_length=1, max_length=50) label: str = Field(..., min_length=1, max_length=100) field_type: FIELD_TYPES required: bool = True options: Optional[list[str]] = None placeholder: Optional[str] = Field(None, max_length=200) help_text: Optional[str] = Field(None, max_length=500) default_value: Optional[str] = None group_name: Optional[str] = Field(None, max_length=100) display_order: int = Field(..., ge=1) validation: Optional[IntakeFieldValidation] = None @field_validator('variable_name') @classmethod def validate_variable_name(cls, v: str) -> str: if not VARIABLE_NAME_PATTERN.match(v): raise ValueError( 'Variable name must start with a lowercase letter and contain only ' 'lowercase letters, numbers, and underscores' ) return v @model_validator(mode='after') def validate_options_for_select(self): if self.field_type in ('select', 'multi_select'): if not self.options or len(self.options) == 0: raise ValueError( f'{self.field_type} fields must have at least one option' ) return self class CategoryInfo(BaseModel): """Embedded category info for tree responses.""" id: UUID name: str slug: str class Config: from_attributes = True class TreeBase(BaseModel): name: str = Field(..., min_length=1, max_length=255) description: Optional[str] = None # Legacy category field - kept for backward compatibility category: Optional[str] = Field(None, max_length=100) class TreeCreate(TreeBase): tree_type: TreeType = Field('troubleshooting', description="Tree type: troubleshooting or procedural") tree_structure: dict[str, Any] = Field(..., description="The decision tree structure in JSON format") intake_form: Optional[list[IntakeFormField]] = Field(None, description="Intake form field definitions (procedural flows only)") is_public: bool = Field(False, description="Make tree visible to all users") is_default: bool = Field(False, description="Mark as a default/system tree (admin only)") status: Literal['draft', 'published'] = Field('published', description="Status: draft or published") category_id: Optional[UUID] = Field(None, description="Category ID from tree_categories table") tags: Optional[list[str]] = Field(None, max_length=10, description="List of tag names to assign") @model_validator(mode='after') def validate_intake_form_unique_variables(self): if self.intake_form: names = [f.variable_name for f in self.intake_form] if len(names) != len(set(names)): raise ValueError('Intake form variable names must be unique') return self class TreeUpdate(BaseModel): name: Optional[str] = Field(None, min_length=1, max_length=255) description: Optional[str] = None category: Optional[str] = Field(None, max_length=100) category_id: Optional[UUID] = None tree_type: Optional[TreeType] = None tree_structure: Optional[dict[str, Any]] = None intake_form: Optional[list[IntakeFormField]] = None is_public: Optional[bool] = None is_active: Optional[bool] = None status: Optional[Literal['draft', 'published']] = None tags: Optional[list[str]] = Field(None, max_length=10, description="List of tag names to assign (replaces existing)") @model_validator(mode='after') def validate_intake_form_unique_variables(self): if self.intake_form: names = [f.variable_name for f in self.intake_form] if len(names) != len(set(names)): raise ValueError('Intake form variable names must be unique') return self class ForkCreate(BaseModel): fork_reason: Optional[str] = Field(None, max_length=255, description="Brief reason for forking") name: Optional[str] = Field(None, min_length=1, max_length=255, description="Name for the fork (defaults to 'Fork of {original name}')") class ForkInfo(BaseModel): """Fork metadata included in tree responses.""" parent_tree_id: Optional[UUID] = None root_tree_id: Optional[UUID] = None fork_reason: Optional[str] = None fork_depth: int = 0 parent_updated_at: Optional[datetime] = None has_parent_updates: bool = False class Config: from_attributes = True class TreeResponse(TreeBase): id: UUID tree_type: str = 'troubleshooting' tree_structure: dict[str, Any] intake_form: Optional[list[dict[str, Any]]] = None author_id: Optional[UUID] = None account_id: Optional[UUID] = None category_id: Optional[UUID] = None category_info: Optional[CategoryInfo] = None tags: list[str] = [] # List of tag names fork_info: Optional[ForkInfo] = None is_active: bool is_public: bool is_default: bool status: str # draft or published version: int created_at: datetime updated_at: datetime usage_count: int class Config: from_attributes = True class TreeListResponse(BaseModel): id: UUID name: str description: Optional[str] = None tree_type: str = 'troubleshooting' category: Optional[str] = None category_id: Optional[UUID] = None category_info: Optional[CategoryInfo] = None tags: list[str] = [] # List of tag names author_id: Optional[UUID] = None author_name: Optional[str] = None # Display name or email of author account_id: Optional[UUID] = None is_active: bool is_public: bool is_default: bool visibility: str = 'team' status: str # draft or published version: int usage_count: int created_at: datetime updated_at: datetime class Config: from_attributes = True # --- Tree Sharing Schemas --- class TreeShareCreate(BaseModel): """Request to create a share token for a tree.""" allow_forking: bool = Field(True, description="Whether recipients can fork this tree") expires_at: Optional[datetime] = Field(None, description="Optional expiration time for the share") class TreeShareResponse(BaseModel): """Response containing share token and URL.""" id: UUID tree_id: UUID share_token: str share_url: str allow_forking: bool created_by: UUID created_at: datetime expires_at: Optional[datetime] = None class Config: from_attributes = True class TreeVisibilityUpdate(BaseModel): """Request to update tree visibility.""" visibility: Literal['private', 'team', 'link', 'public'] = Field(..., description="Visibility level") class SharedTreeResponse(TreeBase): """Public response for shared trees (minimal info).""" id: UUID tree_structure: dict[str, Any] category: Optional[str] = None tags: list[str] = [] version: int allow_forking: bool # From share token created_at: datetime updated_at: datetime class Config: from_attributes = True # --- Tree Validation Schemas --- class ValidationError(BaseModel): """Individual validation error.""" field: str message: str class TreeValidationResponse(BaseModel): """Response for tree validation endpoint.""" can_publish: bool errors: list[ValidationError] = [] # --- Pinned Flows Schemas --- class PinnedFlowResponse(BaseModel): """A pinned flow in the sidebar.""" id: UUID tree_id: UUID tree_name: str tree_type: str category_emoji: Optional[str] = None category_name: Optional[str] = None pinned_at: datetime display_order: int class Config: from_attributes = True class PinnedFlowsListResponse(BaseModel): """List of pinned flows.""" items: list[PinnedFlowResponse] count: int class PinnedFlowReorderItem(BaseModel): """Single item in a reorder request.""" tree_id: UUID display_order: int class PinnedFlowReorderRequest(BaseModel): """Request to reorder pinned flows.""" order: list[PinnedFlowReorderItem]