"""Pydantic schemas for RecipeTask and RecipeSubtask operations.""" import re from typing import Any, Literal, Optional from pydantic import BaseModel, ConfigDict, Field, model_validator from src.backend.models.orm.task import DEFAULT_LOOP_TASK_TYPES, TASK_TYPES # Kept in step with the ORM enum so the two cannot drift apart. TaskType = Literal[TASK_TYPES] def validate_tolerance_order( ltl: Optional[float], lwl: Optional[float], nominal: Optional[float], uwl: Optional[float], utl: Optional[float], ) -> None: """ltl <= lwl <= nominal <= uwl <= utl, checked only across the values set. A one-sided tolerance - only utl, say - is not an error: what is wrong is limits that invert the sequence they do share, like utl landing below ltl. Shared by SubtaskCreate's own validator and by the router's partial-update check (SubtaskUpdate has no fixed set of fields to compare). """ ordered = [ ("ltl", ltl), ("lwl", lwl), ("nominal", nominal), ("uwl", uwl), ("utl", utl), ] present = [(name, value) for name, value in ordered if value is not None] for (prev_name, prev), (name, value) in zip(present, present[1:]): if prev > value: raise ValueError( f"Tolerance limits out of order: {prev_name}={prev} > {name}={value}" ) # #55: campi extra di un task 'note' MAX_NOTE_FIELDS = 20 _FIELD_KEY_RE = re.compile(r"^[a-z0-9_]{1,40}$") NOTE_FIELD_TYPES = ("text", "number") def normalize_note_fields(fields: Optional[list]) -> Optional[list[dict]]: """Valida e normalizza `fields` di un task 'note' (#55); lista vuota -> None. Solleva ValueError con un messaggio leggibile (il router lo trasforma in 422). """ if not fields: return None if len(fields) > MAX_NOTE_FIELDS: raise ValueError(f"Al massimo {MAX_NOTE_FIELDS} campi per task") out: list[dict] = [] seen: set[str] = set() for i, raw in enumerate(fields, start=1): if not isinstance(raw, dict): raise ValueError(f"Campo {i}: formato non valido") key = raw.get("key") if not isinstance(key, str) or not _FIELD_KEY_RE.match(key): raise ValueError( f"Campo {i}: chiave non valida (solo a-z, 0-9, _, da 1 a 40 caratteri)" ) if key in seen: raise ValueError(f"Campo {i}: chiave '{key}' duplicata") seen.add(key) label = raw.get("label") label = label.strip() if isinstance(label, str) else "" if not 1 <= len(label) <= 100: raise ValueError(f"Campo '{key}': etichetta da 1 a 100 caratteri") ftype = raw.get("type") if ftype not in NOTE_FIELD_TYPES: raise ValueError(f"Campo '{key}': tipo deve essere text o number") default = raw.get("default") if default is not None: if not isinstance(default, str): raise ValueError(f"Campo '{key}': il valore predefinito deve essere testo") if default == "": default = None if default is not None and ftype == "number": try: float(default.replace(",", ".")) except ValueError: raise ValueError(f"Campo '{key}': il valore predefinito non e' un numero") attachment = raw.get("attachment", False) if not isinstance(attachment, bool): raise ValueError(f"Campo '{key}': attachment deve essere true o false") out.append({ "key": key, "label": label, "type": ftype, "default": default, "attachment": attachment, }) return out def check_task_loop_options(task_type: str, at_close: bool, fields: Optional[list]) -> None: """at_close e fields sono ammessi solo per i task 'note' (#53 #55).""" if task_type != "note": if at_close: raise ValueError("'Alla chiusura' e' ammesso solo per i task di tipo note") if fields: raise ValueError("I campi extra sono ammessi solo per i task di tipo note") def default_in_loop(task_type: str) -> bool: """in_loop quando non dichiarato: misure e check si', gli altri no (#53).""" return task_type in DEFAULT_LOOP_TASK_TYPES class SubtaskCreate(BaseModel): """Schema for creating a subtask.""" marker_number: int = Field(..., ge=1) description: str = Field(..., min_length=1, max_length=500) measurement_type: Optional[str] = Field(None, max_length=100) nominal: Optional[float] = None utl: Optional[float] = None uwl: Optional[float] = None lwl: Optional[float] = None ltl: Optional[float] = None unit: str = Field("mm", max_length=20) image_path: Optional[str] = Field(None, max_length=500) vision_output: Optional[str] = None @model_validator(mode="after") def _check_tolerance_order(self) -> "SubtaskCreate": validate_tolerance_order(self.ltl, self.lwl, self.nominal, self.uwl, self.utl) return self class SubtaskUpdate(BaseModel): """Schema for updating a subtask.""" marker_number: Optional[int] = Field(None, ge=1) description: Optional[str] = Field(None, min_length=1, max_length=500) measurement_type: Optional[str] = Field(None, max_length=100) nominal: Optional[float] = None utl: Optional[float] = None uwl: Optional[float] = None lwl: Optional[float] = None ltl: Optional[float] = None unit: Optional[str] = Field(None, max_length=20) image_path: Optional[str] = Field(None, max_length=500) vision_output: Optional[str] = None class SubtaskResponse(BaseModel): """Schema for subtask response.""" model_config = ConfigDict(from_attributes=True) id: int task_id: int marker_number: int description: str measurement_type: Optional[str] = None nominal: Optional[float] = None utl: Optional[float] = None uwl: Optional[float] = None lwl: Optional[float] = None ltl: Optional[float] = None unit: str image_path: Optional[str] = None vision_output: Optional[str] = None class TaskCreate(BaseModel): """Schema for creating a task.""" title: str = Field(..., min_length=1, max_length=255) task_type: TaskType = "note" directive: Optional[str] = None description: Optional[str] = None file_path: Optional[str] = Field(None, max_length=500) file_type: Optional[str] = Field(None, pattern="^(image|pdf)$") annotations_json: Optional[dict[str, Any]] = None vision_json: Optional[dict] = None # #53: assente = default dal tipo (misure e check nel ciclo) in_loop: Optional[bool] = None at_close: bool = False fields: Optional[list[dict[str, Any]]] = None subtasks: list[SubtaskCreate] = [] class TaskUpdate(BaseModel): """Schema for updating a task.""" title: Optional[str] = Field(None, min_length=1, max_length=255) task_type: Optional[TaskType] = None directive: Optional[str] = None description: Optional[str] = None file_path: Optional[str] = Field(None, max_length=500) file_type: Optional[str] = Field(None, pattern="^(image|pdf)$") annotations_json: Optional[dict[str, Any]] = None vision_json: Optional[dict] = None in_loop: Optional[bool] = None at_close: Optional[bool] = None fields: Optional[list[dict[str, Any]]] = None class TaskResponse(BaseModel): """Schema for task response.""" model_config = ConfigDict(from_attributes=True) id: int version_id: int recipe_id: Optional[int] = None order_index: int task_type: str title: str directive: Optional[str] = None description: Optional[str] = None file_path: Optional[str] = None file_type: Optional[str] = None annotations_json: Optional[dict[str, Any]] = None vision_json: Optional[dict] = None in_loop: bool = False at_close: bool = False fields: Optional[list[dict[str, Any]]] = None subtasks: list[SubtaskResponse] = [] class TaskReorderRequest(BaseModel): """Schema for reordering tasks.""" task_ids: list[int] = Field(..., min_length=1)