import uuid from datetime import datetime, timedelta, timezone from typing import Annotated, Literal from pydantic import BaseModel, ConfigDict, Field, StringConstraints, computed_field, model_validator from app.core.expr import expression_error from app.schemas.actions import ActionStep, check_flow from app.schemas.master import UserPickerType from app.schemas.reporting import DatasetOperator from app.schemas.styles import CustomCss, ElementStyles from app.schemas.validation_rules import FormValidationRule, FormValidationRuleType # noqa: F401 — re-exported PublishStatus = Literal["draft", "published"] InstanceStatus = Literal["in_progress", "completed", "cancelled", "escalated"] # The Form Designer's field palette, grouped to match its UI categories. # "dropdown" is the pre-existing type id for what the palette now labels # "Select" — kept as-is rather than renamed, since saved forms already use # it and JSONB has no migration step to force. FormFieldType = Literal[ # INPUT — "phone" is "+ " text (default_country picks # the code shown first); "address" a dict of ADDRESS_PARTS; "barcode" # text read by the camera (or typed); "autonumber" text the server gives # once, on submit, in autonumber_format (like a datasource's autonumber). "text", "textarea", "password", "email", "number", "currency", "percentage", "url", "phone", "address", "barcode", "autonumber", # SELECTION — "toggle" is a switch holding true/false like a checkbox; # "consent" a checkbox beside consent_text that, when required, must be # ticked. "matrix" asks each of matrix_rows one of `options` (a Likert # grid), stored as {row: option}; "ranking" orders all of `options`, # stored as the ordered list. "checkbox", "toggle", "consent", "radio", "dropdown", "multiselect", "autocomplete", "combobox", "treeselect", "lookup", "matrix", "ranking", "user", # dropdown live-linked to this tenant's system Users — see app/crud/user_picker.py # DATE_TIME "date", "time", "datetime", "daterange", "duration", # FILE — "multi_attachment" holds a list of uploaded files' ids (dropped # or picked several at a time), within max_files / max_file_size_mb. # "camera" is a photo taken with the device's camera, kept like an image upload. "file_upload", "image_upload", "attachment", "multi_attachment", "signature", "camera", # CONTENT "rich_text", "markdown", "html", "code", "json", # ADVANCED "slider", "rating", "tags", "color_picker", "formula", "dynamic_grid", # Repeating line items: a list of rows, each with subform_fields (full # field types, their own validation and per-row formulas). The parent's # formulas can total them, e.g. SUM(items.amount). "subform", # Another published Form placed inside this one ("embedded_form_id"): its # fields are filled here, and on submit its answers are saved as that # form's own submission — this field keeps the submission's id. See # app/routers/form_submissions.py::_save_embedded. "embedded_form", # LOCATION — a point picked on a map, stored as "lat,lng" text (see # app/core/form_validation.py::parse_geolocation) "geolocation", # DATA DISPLAY — read-only ("label") or invisible ("hidden"); unlike # LAYOUT below, both still take a real value via values[name] (typically # filled by a Form's prefill config, see FormPrefillConfig, rather than # direct user input) — "label" renders it as static text, "hidden" # doesn't render at all but still rides along into the submission. "label", "hidden", # LAYOUT — the field list stays flat; a container marker ("section", # "panel", "tabs", "accordion", "row", "wizard") groups every field after # it up to the next container marker or "group_end", and "column" splits # the current group into side-by-side columns. Adjacent "tabs"/ # "accordion" markers form one tab strip / accordion. See # frontend/src/lib/formLayout.ts and app/core/form_validation.py's # layout_owners (a hidden container hides its fields too). "section", "panel", "tabs", "accordion", "wizard", "row", "column", "group_end", # ACTIONS — carries no value; see FormFieldDef.button_kind/button_action below. "button", ] # Mirrors the frontend's FORM_CHOICE_FIELD_TYPES (frontend/src/lib/types.ts) — # fields that resolve their option list from either static options or a # live-linked Master, and so are eligible for cascade_parent_field/ # cascade_match_field below. _CHOICE_FIELD_TYPES = {"radio", "dropdown", "multiselect", "autocomplete", "combobox", "treeselect", "lookup"} # An "address" field's parts, in the order the form shows them. ADDRESS_PARTS = ("line1", "line2", "city", "state", "postal_code", "country") # ---- Workflows ------------------------------------------------------------- WorkflowConditionOperator = Literal["equals", "not_equals", "gt", "gte", "lt", "lte", "in"] WorkflowNotificationTrigger = Literal["on_enter", "on_complete"] ParamType = Literal["int", "number", "string", "boolean", "list", "object", "any"] # same set as Tasks (app/crud/params.py) class WorkflowInputParam(BaseModel): """Same generic, runtime-enforced param contract as a Task node's (app/schemas/automation.py::NodeInputParam) — see app/crud/params.py, the one validator both engines share. A workflow instance's context (WorkflowInstance.context) is what these are resolved against: an upstream step's declared output_params, plus whatever the caller passed starting an instance, is the only way values reach a later step.""" name: str type: ParamType source_field: str required: bool = True class WorkflowOutputParam(BaseModel): name: str # also the context key this output is written to type: ParamType source: str = "" # path inside the step's result; "*" = all of it (app/crud/params.py) class WorkflowStepCondition(BaseModel): driver_field: str operator: WorkflowConditionOperator value: object class WorkflowStepTransition(BaseModel): to: str condition: WorkflowStepCondition | None = None # None on a non-decision step's single transition class WorkflowStepNotification(BaseModel): connector_id: uuid.UUID trigger: WorkflowNotificationTrigger message: str = "" # email_smtp connectors only: names a field on the most recent Form # submission tied to this workflow instance whose value is the "To" # address (e.g. a client's email captured on an order form) — see # app/crud/notifications.py's _resolve_recipient. recipient_field: str | None = None # Who gets the in-app notification: every active member of these roles # (internal or Public Portal scope) plus these individual users (internal # or vendor accounts), resolved when the notification fires — each gets # their own row/read state. Both empty = the original tenant-wide # broadcast to internal users. For an email_smtp connector, each resolved # user's email address is also sent the email, alongside recipient_field. recipient_role_ids: list[uuid.UUID] = [] recipient_user_ids: list[uuid.UUID] = [] WorkflowStepTargetType = Literal["master", "form"] # What each step type does (app/crud/workflow_engine.py runs them): # start — where an instance begins; moves straight on unless it names # approvers, then it waits for them like a task. # task, approval — wait for a person: the approvers (roles, users, or the # user named by assignee_field), any one of them / all / a # majority (approval_mode); SLA, reminder and escalation. # decision — routes itself: the first branch whose condition holds (a # branch with no condition is the "otherwise"). With approvers # it's a person's choice among the branches that hold instead. # parallel — starts every outgoing branch at once (a branch with a # condition only if it holds). # join — waits for the branches of its parallel split: all of them, # or the first to arrive (join_mode). # timer — waits a number of hours, or until the date in a field. # subprocess — starts another workflow (one per list item with for_each) # and waits for it, mapping values in and out. # service — runs a Task in the background, with retries; its outputs # land in the context. A failure for good is an incident. # end — the branch is done; the instance is done when every branch # is (or at once, with terminate). WorkflowStepType = Literal["start", "task", "approval", "decision", "end", "parallel", "join", "timer", "subprocess", "service"] HUMAN_STEP_TYPES = frozenset({"task", "approval"}) ApprovalMode = Literal["any", "all", "majority"] class WorkflowEscalation(BaseModel): """What happens when a person's step passes its SLA (sla_hours).""" # notify: tell these people (and the assignees); reassign: the step now # waits on these people instead; route: move on to route_to. action: Literal["notify", "reassign", "route"] = "notify" role_ids: list[uuid.UUID] = [] user_ids: list[uuid.UUID] = [] route_to: str | None = None class WorkflowTimer(BaseModel): # duration: wait `hours`; until_field: wait until the date/time in context[field]. mode: Literal["duration", "until_field"] = "duration" hours: float | None = Field(default=None, ge=0, le=24 * 366) field: str | None = None class WorkflowSubprocess(BaseModel): workflow_id: uuid.UUID | None = None # child context key -> parent context key it's copied from (a blank value: the same name). input_map: dict[str, str] = {} # parent context key -> child context key it's copied back from, when the child completes. output_map: dict[str, str] = {} # A parent context key holding a list: one child per item, each with # "item" (and "item_index") in its context; outputs come back as lists. for_each: str | None = None # False: start it and move on at once. wait: bool = True class WorkflowService(BaseModel): task_id: uuid.UUID | None = None # Extra attempts after a failure (a Task may send email or call out, so none by default). retries: int = Field(default=0, ge=0, le=10) class WorkflowSettings(BaseModel): # Step SLAs, reminders and timers count Monday–Friday only. business_days: bool = False class WorkflowStepDef(BaseModel): key: str type: WorkflowStepType label: str approver_roles: list[str] = [] approver_users: list[uuid.UUID] = [] # A context key holding a user id (or a list of them) who must act — # e.g. the requester's manager. Adds to the approvers above. assignee_field: str | None = None approval_mode: ApprovalMode = "any" sla_hours: float | None = Field(default=None, ge=0) # Remind the assignees this many hours before the SLA is due. remind_before_hours: float | None = Field(default=None, ge=0) escalation: WorkflowEscalation | None = None join_mode: Literal["all", "any"] = "all" timer: WorkflowTimer | None = None subprocess: WorkflowSubprocess | None = None service: WorkflowService | None = None # An end step that ends the whole instance, cancelling any other branches. terminate: bool = False next_steps: list[WorkflowStepTransition] = [] notifications: list[WorkflowStepNotification] = [] # What this step reads from / writes to the instance's context — see # WorkflowInputParam above. Checked on start (the Start step's inputs, # against the caller-supplied context) and on each advance (the leaving # step's outputs, then the entering step's inputs) — app/routers/ # workflow_instances.py. input_params: list[WorkflowInputParam] = [] output_params: list[WorkflowOutputParam] = [] # What this step is "about" — a datasource to view/list (e.g. "view # order requests") or a form to fill in — same target_type/target_id # convention Menu uses (see app/schemas/navigation.py's MenuCreate). target_type: WorkflowStepTargetType | None = None target_id: uuid.UUID | None = None # Free-form canvas placement (Workflow Designer) — purely presentational, # never read by the instance engine. position_x: float = 0 position_y: float = 0 class WorkflowCreate(BaseModel): name: str step_definitions: list[WorkflowStepDef] = [] settings: WorkflowSettings = WorkflowSettings() class WorkflowUpdate(BaseModel): name: str | None = None step_definitions: list[WorkflowStepDef] | None = None settings: WorkflowSettings | None = None class WorkflowDuplicate(BaseModel): # Blank/omitted -> " (copy)". name: str | None = Field(default=None, max_length=200) class WorkflowRead(BaseModel): model_config = ConfigDict(from_attributes=True) workflow_id: uuid.UUID version: int tenant_id: uuid.UUID name: str step_definitions: list[dict] settings: dict = {} status: PublishStatus is_current: bool created_by: uuid.UUID | None created_at: datetime updated_at: datetime # ---- Forms ------------------------------------------------------------- # A form's OnSave/OnSubmit/OnCancel lifecycle hooks (Task Designer Studio # integration). Each event runs at most one action: # - run_task / trigger_workflow: real backend side effects (POST # /tasks/{id}/run, POST /workflow-instances), given whatever context # (e.g. the submitted field values) the frontend has on hand. On_submit # specifically also gets the just-created FormSubmission's own id, under # context key "recordId" (same name the ?recordId= query-param # convention uses elsewhere) — the only way a run_task's data_update # node can target that exact submission later, since match_field "id" # (app/schemas/automation.py::TaskNodeDef) has no field VALUE to look # for until this event fires. # - navigate / show_message: frontend-only reactions, same target-entity # convention Page Designer's PageAction already uses # on_load: when the form opens (after prefill). FormEventTrigger = Literal["on_save", "on_submit", "on_cancel", "on_load"] # "none" is a caption-only lifecycle event — it carries a button_label (see # FormEventAction) but runs nothing. # run_flow: an ordered list of steps (app/schemas/actions.py) instead of one action. FormEventActionType = Literal["none", "run_task", "trigger_workflow", "navigate", "show_message", "open_external_url", "run_flow"] FormNavigateTargetType = Literal["page", "master", "form", "dashboard"] class FormActionParam(BaseModel): """One extra query parameter a form's navigate action passes to its target (?name=value — a page parameter there, or a form's prefill). `value` is text or a template: "{{field_name}}" (this form's values), "{{recordId}}" (the submission just saved), a URL parameter of the page the form is on, "{{current_user.id}}", "{{session.key}}" — resolved in the browser; an empty result is left out. Same shape as navigation.PageActionParam.""" name: str = Field(min_length=1, max_length=100) value: str = Field(default="", max_length=2000) class FormActionDef(BaseModel): """The action itself, independent of what fires it — a form's on_save/ on_submit/on_cancel lifecycle hooks (FormEventAction below, adds `trigger`) and a canvas "button" field (FormFieldDef.button_action) both fire one of these.""" action_type: FormEventActionType task_id: uuid.UUID | None = None workflow_id: uuid.UUID | None = None target_type: FormNavigateTargetType | None = None target_id: uuid.UUID | None = None message: str | None = None # open_external_url — sends the browser off-site (a new tab); can be # "{{field_name}}", rendered against the form's own submitted values — # e.g. a hidden field the form itself collected, or one prefilled by a # Master-linked lookup (see FormRunner.tsx's runEvent for the render # step). Only one event fires per trigger, so this can't directly chain # off a same-trigger run_task's result — a Page data_list's row-click # open_external_url (PageAction.url), rendered against the clicked # row's own stored field values, is the supported way to redirect to a # value a prior step (e.g. payment_initiate) wrote onto a record. url: str | None = None # navigate only — append the just-created FormSubmission's own id as # ?recordId= on the target path (same convention as PageAction.pass_row_id # for a data_list row-click). Meaningless for on_save/on_cancel/ # action_only button clicks, since no submission exists yet at that point. pass_record_id: bool = False # navigate only — more query parameters for the target; see FormActionParam. params: list[FormActionParam] = [] steps: list[ActionStep] = [] # run_flow only @model_validator(mode="after") def _check_action_fields(self) -> "FormActionDef": if self.action_type == "run_flow": if not self.steps: raise ValueError("run_flow actions need at least one step") check_flow(self.steps) if self.action_type == "run_task" and not self.task_id: raise ValueError("run_task actions require task_id") if self.action_type == "trigger_workflow" and not self.workflow_id: raise ValueError("trigger_workflow actions require workflow_id") if self.action_type == "navigate" and (not self.target_type or not self.target_id): raise ValueError("navigate actions require target_type and target_id") if self.action_type == "show_message" and not self.message: raise ValueError("show_message actions require message") if self.action_type == "open_external_url" and not self.url: raise ValueError("open_external_url actions require url") return self class FormEventAction(FormActionDef): trigger: FormEventTrigger # The caption of the button that fires this trigger — Submit's for # on_save, Cancel's for on_cancel (on_submit fires without a button of its # own, so it ignores this). Blank = the default caption. Kept on the # event, rather than a new Form column, so it needs no schema change and # versions/copies along with `events`. button_label: str | None = None class FormVisibilityRule(BaseModel): driver_field: str operator: Literal["equals", "not_equals", "gt", "gte", "lt", "lte", "in"] value: object class GridColumnDef(BaseModel): name: str label: str type: Literal["text", "number", "date", "checkbox"] FormButtonKind = Literal["action_only", "submit_and_action"] class FormOptionFilter(BaseModel): """Narrows a choice field's options at runtime. `value` may hold "{{name}}" tokens resolved in the browser against the form's context — a URL query parameter, "current_user.", or "session." (the application context a Page's on_load task stored). `field` names a field on the linked datasource ("id" = the record's own id); static options are matched on the option text itself and ignore it. Client-side only, like cascade_parent_field — see frontend/src/lib/optionFilters.ts.""" field: str = "" operator: DatasetOperator = "equals" value: str = "" # When the resolved value is blank (or a token it uses has no value): # True skips this filter (all options stay); False (default) leaves no # options, so a missing context value never widens the list. skip_if_empty: bool = False FormFieldAccess = Literal["read_only", "hidden"] class WizardJump(BaseModel): """On a "wizard" step: when `when` (an expression over the form's values) is true as the person leaves this step, go to the step opened by the wizard field named `goto` — or "end" to make this the last step. The first rule that holds wins; none = the next step. Steps jumped over don't count (their required fields aren't asked for) — the same path is worked out on the server (app/core/form_validation.py::wizard_unreached).""" when: str = Field(min_length=1, max_length=2000) goto: str = Field(min_length=1, max_length=100) @model_validator(mode="after") def _check(self) -> "WizardJump": error = expression_error(self.when) if error: raise ValueError(f"wizard jump condition — {error}") return self class FormFieldAccessRule(BaseModel): """Narrows who may change or see one field (app/crud/field_access.py): for holders of any of `role_ids` (empty = everyone), while the submission's workflow sits at one of `steps` (empty = always), the field is read-only or hidden. The most restrictive matching rule wins. Enforced on the server: a hidden field is left out of everything the person reads, and neither kind of field can be changed by them when editing.""" role_ids: list[uuid.UUID] = [] steps: list[str] = [] access: FormFieldAccess = "read_only" # Not allowed inside a line item: another subform, layout markers, buttons. # Fields that can hold another record's id and say whose (link_source). _LINKABLE_TYPES = {"hidden", "text", "label"} _NOT_IN_SUBFORM = {"subform", "embedded_form", "autonumber", "section", "panel", "tabs", "accordion", "wizard", "row", "column", "group_end", "button", "dynamic_grid"} class FormFieldDef(BaseModel): name: str type: FormFieldType label: str required: bool = False options: list[str] | None = None # static options — choice-family types (select/radio/multiselect/...) master_id: uuid.UUID | None = None # live-linked choice source (FR-FRM-004) master_display_field: str | None = None # which field of that master's schema labels each option # Choice options from a Data Query's rows (app/crud/data_queries.py) — # the stored value is options_value_field, shown as options_label_field. options_query_id: uuid.UUID | None = None options_value_field: str | None = None options_label_field: str | None = None visibility_rules: list[FormVisibilityRule] = [] # label/placeholder may hold "{{name}}" tokens, and default_value is a # template too — all resolved in the browser against the form's context: # a URL/page parameter, "session." or "current_user." (see # frontend/src/lib/contextTemplates.ts). A default fills the field when # the form opens, like prefill, and never overwrites a typed value. default_value: str | None = None placeholder: str | None = None min: float | None = None max: float | None = None step: float | None = None rows: int | None = None # textarea-family height accept: str | None = None # file-family accept filter, e.g. "image/*" # Grid layout. On a container marker (section/panel/tabs/accordion/ # wizard/row): how many columns its fields flow into, left to right and # row by row (1 = stacked, the default). On any field: how many of those # columns it takes (col_span >= the group's columns = the whole row). # Narrow screens always stack to one column (frontend FormLayout.tsx). layout_columns: int | None = Field(default=None, ge=1, le=4) col_span: int | None = Field(default=None, ge=1, le=4) # radio / checkbox-list choices side by side instead of one per line. options_inline: bool = False # "multi_attachment" only — how many files, and how large each may be. max_files: int | None = Field(default=None, ge=1, le=100) max_file_size_mb: float | None = Field(default=None, gt=0, le=1024) rating_max: int | None = None grid_columns: list[GridColumnDef] | None = None # "dynamic_grid" only # "formula" only — an expression (app/core/expr.py) over the other # fields, e.g. "{quantity} * {unit_price}"; computed live in the browser # and again on the server when saved. formula_expression: str | None = None # Shown only while this expression is true (on top of visibility_rules), # e.g. "amount > 1000 and HASROLE('Approver')". Same language and same # result in the browser and on the server. visible_when: str | None = None user_type: UserPickerType | None = None # "user" only — which pool of accounts this field draws from user_role_id: uuid.UUID | None = None # "user" only — further restrict the pool to holders of one Role # choice-family only — cascading/dependent filtering, client-side # convenience only (not enforced on submission): cascade_parent_field # names another field ON THIS SAME FORM whose currently-entered value # narrows this field's options; cascade_match_field names the field on # master_id's own schema whose value must equal the parent's value. See # frontend/src/components/FormFieldRenderer.tsx's useChoiceOptions. cascade_parent_field: str | None = None cascade_match_field: str | None = None # choice-family only — see FormOptionFilter. option_filters: list[FormOptionFilter] = [] # "button" only — see FormActionDef above. action_only fires button_action # without submitting; submit_and_action submits the form first (same flow # as the form's own Submit button) and then fires button_action. button_kind: FormButtonKind | None = None button_action: FormActionDef | None = None # Shown instead of the default "This field is required." when blank. required_message: str | None = None validation_rules: list[FormValidationRule] = [] # Builds a database index on this field's values (app/crud/field_indexes.py) # so filtering, ranges and lookups on it stay fast on very large data. indexed: bool = False # Keeps the search box fast on very large data (app/crud/field_indexes.py). searchable: bool = False # Who may change or see this field — see FormFieldAccessRule. access_rules: list[FormFieldAccessRule] = [] # Steps run whenever this field's value changes (app/schemas/actions.py) # — e.g. set_field to fill another field, or show_message. on_change: list[ActionStep] = [] # "subform" only — the fields of each line item. Row formulas see the # row's fields by name and the form's as parent.. subform_fields: list["FormFieldDef"] | None = None # "embedded_form" only — the Form whose fields are filled in this one. embedded_form_id: uuid.UUID | None = None # "autocomplete" only — several choices, stored as one comma-separated # text ("Kochi, Thrissur"); each must be one of the options. allow_multiple: bool = False # "label" and "hidden" — what it shows/holds (and saves): "value" whatever prefill or a # flow put there; "text" label_text; "record" one field # (label_display_field) of the first record of a datasource or form # (label_source_type/label_source_id) whose label_match_field equals # label_match_value. label_text and label_match_value are templates: # "{{key}}" a page/URL parameter, session.* or current_user.* value, # "{field}" another field of this form (so the label follows it). label_source: Literal["value", "text", "record"] = "value" label_text: str | None = Field(default=None, max_length=2000) label_source_type: Literal["master", "form"] | None = None label_source_id: uuid.UUID | None = None label_display_field: str | None = None label_match_field: str | None = None label_match_value: str | None = Field(default=None, max_length=500) # "label" only — its box's height in px (a longer text scrolls inside); # blank = fit to content. label_height: int | None = Field(default=None, ge=24, le=2000) # "hidden" / "text" / "label" holding another record's id (e.g. an # application's submission id): which datasource or form it points at # ("master:" / "form:"), and which of that record's fields names # it. Pages can then show the linked record's fields as columns # ("application_id.applicant_name") and its name instead of the id. link_source: str | None = Field(default=None, pattern=r"^(master|form):[0-9a-fA-F-]{36}$") link_display_field: str | None = None min_rows: int | None = Field(default=None, ge=0) max_rows: int | None = Field(default=None, ge=1) # "consent" only — what the person agrees to (plain text; [text](https://…) # makes a link). consent_text: str | None = Field(default=None, max_length=4000) # "matrix" only — the statements, each answered with one of `options`. matrix_rows: list[str] | None = None # "phone" only — the dial code offered first, e.g. "+91". default_country: str | None = Field(default=None, pattern=r"^\+\d{1,4}$") # "autonumber" only — e.g. "RTI/{YYYY}/{SEQ:5}"; {SEQ} the running number, # {YYYY} {YY} {MM} {DD} the submit date. Reset: start again each year/month. autonumber_format: str | None = Field(default=None, max_length=100) autonumber_reset: Literal["never", "yearly", "monthly"] = "never" # "wizard" only — where Next goes from this step; see WizardJump. next_step_rules: list[WizardJump] = [] # "email" / "phone" only — a one-time code is sent to what's typed in, # and the form can't be submitted until it's entered (app/crud/ # form_verification.py). Needs the workspace's email / SMS connector. require_verification: bool = False # Form Designer › Style — this field's look (its box, label, input, # button, a line-item table's th/tr/td) and free CSS scoped to it. styles: ElementStyles | None = None custom_css: CustomCss | None = None @model_validator(mode="after") def _check_user_fields(self) -> "FormFieldDef": for label, expression in (("formula", self.formula_expression), ("show when", self.visible_when)): if expression and expression.strip(): error = expression_error(expression) if error: raise ValueError(f"field '{self.name}': {label} expression — {error}") self.visible_when = (self.visible_when or "").strip() or None if self.type == "user": if self.user_type is None: raise ValueError("user fields need user_type") elif self.user_type is not None or self.user_role_id is not None: raise ValueError("user_type/user_role_id are only meaningful for type 'user'") return self @model_validator(mode="after") def _check_subform(self) -> "FormFieldDef": if self.type == "subform": if not self.subform_fields: raise ValueError(f"field '{self.name}': line items need at least one field") bad = [f.name for f in self.subform_fields if f.type in _NOT_IN_SUBFORM] if bad: raise ValueError(f"field '{self.name}': {', '.join(bad)} can't be used inside line items") names = [f.name for f in self.subform_fields] if len(set(names)) != len(names): raise ValueError(f"field '{self.name}': line item field names must be unique") elif self.subform_fields: raise ValueError("subform_fields is only meaningful for type 'subform'") return self @model_validator(mode="after") def _check_embedded_form(self) -> "FormFieldDef": if self.type == "embedded_form" and self.embedded_form_id is None: raise ValueError(f"field '{self.name}': choose the form to embed") if self.type != "embedded_form" and self.embedded_form_id is not None: raise ValueError("embedded_form_id is only meaningful for type 'embedded_form'") if self.allow_multiple and self.type != "autocomplete": raise ValueError("allow_multiple is only meaningful for type 'autocomplete'") if self.type not in ("label", "hidden") and (self.label_source != "value" or self.label_source_id is not None): raise ValueError("label_source settings are only meaningful for types 'label' and 'hidden'") if self.type != "label" and self.label_height is not None: raise ValueError("label_height is only meaningful for type 'label'") if self.link_source and self.type not in _LINKABLE_TYPES: raise ValueError(f"field '{self.name}': only hidden, text and label fields can link to another record") if self.link_display_field and not self.link_source: raise ValueError(f"field '{self.name}': link_display_field needs link_source") if self.label_source == "record": if not (self.label_source_type and self.label_source_id and self.label_display_field and self.label_match_field): raise ValueError(f"field '{self.name}': choose the datasource or form, the field to show, and the field to match on") return self @model_validator(mode="after") def _check_query_options(self) -> "FormFieldDef": if self.options_query_id is not None: if self.type not in _CHOICE_FIELD_TYPES: raise ValueError("options_query_id is only meaningful for choice-family field types") if self.master_id is not None: raise ValueError(f"field '{self.name}': choose options from a datasource or a data query, not both") if not self.options_value_field: raise ValueError(f"field '{self.name}': choose which query column holds each option's value") return self @model_validator(mode="after") def _check_cascade_fields(self) -> "FormFieldDef": if bool(self.cascade_parent_field) != bool(self.cascade_match_field): raise ValueError("cascade_parent_field and cascade_match_field must be set together") if self.cascade_parent_field and self.type not in _CHOICE_FIELD_TYPES: raise ValueError("cascade_parent_field/cascade_match_field are only meaningful for choice-family field types") if self.cascade_parent_field == self.name: raise ValueError("a field cannot cascade from itself") if self.option_filters and self.type not in _CHOICE_FIELD_TYPES: raise ValueError("option_filters are only meaningful for choice-family field types") return self @model_validator(mode="after") def _check_button_fields(self) -> "FormFieldDef": if self.type == "button": if self.button_kind is None: raise ValueError("button fields need button_kind") if self.button_action is None: raise ValueError("button fields need button_action") elif self.button_kind is not None or self.button_action is not None: raise ValueError("button_kind/button_action are only meaningful for type 'button'") return self @model_validator(mode="after") def _check_new_controls(self) -> "FormFieldDef": if self.type == "matrix": rows = [r.strip() for r in self.matrix_rows or [] if r and r.strip()] if not rows or not self.options: raise ValueError(f"field '{self.name}': a matrix needs its statements (rows) and the answers (options)") if len(set(rows)) != len(rows): raise ValueError(f"field '{self.name}': matrix statements must be different") self.matrix_rows = rows elif self.matrix_rows: raise ValueError("matrix_rows is only meaningful for type 'matrix'") if self.type == "ranking" and len(self.options or []) < 2: raise ValueError(f"field '{self.name}': ranking needs at least two options to order") if self.consent_text is not None and self.type != "consent": raise ValueError("consent_text is only meaningful for type 'consent'") if self.default_country is not None and self.type != "phone": raise ValueError("default_country is only meaningful for type 'phone'") if self.type == "autonumber": self.autonumber_format = (self.autonumber_format or "").strip() or "{SEQ}" if "{SEQ" not in self.autonumber_format: raise ValueError(f"field '{self.name}': the number format needs {{SEQ}} (the running number)") elif self.autonumber_format is not None: raise ValueError("autonumber_format is only meaningful for type 'autonumber'") if self.require_verification and self.type not in ("email", "phone"): raise ValueError(f"field '{self.name}': only email and phone fields can be verified with a code") if self.next_step_rules and self.type != "wizard": raise ValueError("next_step_rules is only meaningful for type 'wizard'") return self FormFieldDef.model_rebuild() class FormPrefillFieldMapping(BaseModel): source_field: str # field name on the datasource record form_field: str # this Form's own field name to fill class FormPrefillContextMapping(BaseModel): # A key of the form's runtime context: a URL query parameter's name # ("employeeId"), "session." (Users → Session data) or # "current_user.". "{{key}}" is accepted too. context_key: str form_field: str class FormPrefillConfig(BaseModel): """Two independent ways to fill this Form's fields on load, wherever it renders (Public Portal standalone runtime, or embedded/opened from a Page) — see frontend's useFormPrefill.ts: - context_mappings copy a context value straight into a field. - A lookup (source_id set) fetches one record via the same POST /datasets/query every Page data grid/widget already uses (source_type/source_id + an equals filter on match_field), matched against the context value named param_name — a URL parameter, "session." or "current_user." — and copies its fields per field_mappings. Never overwrites what the user already typed.""" source_type: Literal["master", "form"] = "master" source_id: uuid.UUID | None = None match_field: str = "" param_name: str = "" field_mappings: list[FormPrefillFieldMapping] = [] context_mappings: list[FormPrefillContextMapping] = [] @model_validator(mode="before") @classmethod def _blank_source_is_none(cls, data): if isinstance(data, dict) and data.get("source_id") == "": data = {**data, "source_id": None} return data def _check_prefill(prefill: FormPrefillConfig | None, field_definitions: list[FormFieldDef] | None) -> None: """Input-only checks (FormCreate/FormUpdate, never FormRead). Mapping targets are only checked when field_definitions are in the same payload — same trust posture as app/schemas/master.py's _check_scope_field.""" if prefill is None: return if prefill.source_id is not None: if not prefill.match_field or not prefill.param_name.strip(): raise ValueError("prefill lookup needs a match field and the context key to match on") if not prefill.field_mappings: raise ValueError("prefill needs at least one field mapping") elif prefill.field_mappings: raise ValueError("prefill field mappings need a datasource to look up") if not prefill.field_mappings and not prefill.context_mappings: raise ValueError("prefill needs at least one field mapping") for m in prefill.context_mappings: if not m.context_key.strip(): raise ValueError(f"prefill context mapping for '{m.form_field}' needs a context key") if field_definitions is None: return names = {f.name for f in field_definitions} for m in [*prefill.field_mappings, *prefill.context_mappings]: if m.form_field not in names: raise ValueError(f"prefill field mapping targets '{m.form_field}', which isn't a field on this Form") def _check_validation_refs(field_definitions: list[FormFieldDef] | None) -> None: """An "equals_field" rule must name another field on the same Form.""" if field_definitions is None: return names = {f.name for f in field_definitions} for f in field_definitions: for rule in f.validation_rules: if rule.type == "equals_field" and (str(rule.value) not in names or str(rule.value) == f.name): raise ValueError(f"field '{f.name}': 'must match' rule names '{rule.value}', which isn't another field on this Form") class FormTestScenario(BaseModel): """Sample values and what should happen to them — run by the Form Designer's Tests and scripts/run_form_scenarios.py (app/crud/ form_scenarios.py) through the same checks as a real submission.""" name: str = Field(min_length=1, max_length=200) values: dict = {} # Role names the test user holds — for HASROLE(...) and "show when". roles: list[str] = [] expect_valid: bool = True # field name -> text the message should contain ("" = any problem). expect_errors: dict[str, str] = {} # field name -> the stored value, formulas included. expect_values: dict = {} class FieldTranslation(BaseModel): label: str | None = Field(default=None, max_length=500) placeholder: str | None = Field(default=None, max_length=500) required_message: str | None = Field(default=None, max_length=500) # static option -> how it reads in this language (the stored value stays the option) options: dict[str, str] = {} class FormTranslation(BaseModel): """One language's wording of a form — shown when a visitor picks that language in the Portal (frontend lib/formTranslations.ts). Anything left out shows as designed.""" title: str | None = Field(default=None, max_length=200) fields: dict[str, FieldTranslation] = {} class FormDocument(BaseModel): """A printable PDF of a submission (app/crud/form_documents.py). `body` is plain text: "{field_name}" is that answer, "{answers}" a table of the answers (`fields`, blank = every answered field), and {form}, {workspace}, {submitted_at} what they say. Blank body = the table.""" enabled: bool = False title: str | None = Field(default=None, max_length=200) body: str | None = Field(default=None, max_length=20000) footer: str | None = Field(default=None, max_length=1000) fields: list[str] = [] class FormReceipt(BaseModel): """An email sent when the form is submitted, to the address in the `to_field` email field — subject and message are templates like the document's body; the PDF goes with it when attach_pdf is on.""" enabled: bool = False to_field: str | None = None subject: str | None = Field(default=None, max_length=300) message: str | None = Field(default=None, max_length=10000) attach_pdf: bool = True class FormSchedule(BaseModel): """When the form takes new submissions (app/crud/form_schedule.py): from opens_at until closes_at (either blank = no limit), unless paused or archived. A closed form still shows its submissions, and a submission returned for correction can still be corrected.""" opens_at: datetime | None = None closes_at: datetime | None = None paused: bool = False # An old program: closed for good, kept for its data. archived: bool = False # Shown instead of the form while it isn't open; blank = a default line. message: str | None = Field(default=None, max_length=1000) @model_validator(mode="after") def _order(self) -> "FormSchedule": if self.opens_at and self.closes_at and self.closes_at <= self.opens_at: raise ValueError("The closing date must be after the opening date.") return self class FormReview(BaseModel): """Officers review submissions (app/crud/application_review.py): each gets a status — submitted, under review, returned for correction, resubmitted, approved, rejected, closed — with remarks and a history. status_field: a form field the status label is copied into, so lists, filters, dashboards and exports can use it like any answer. notify_field: an email or phone field told of each status change. Public forms: status_lookup lets applicants check their status with the reference number (an auto-number field) and a code sent to the email or phone they gave (lookup_verify_field); a returned one can be corrected there within correction_days.""" enabled: bool = False status_field: str | None = None notify_field: str | None = None status_lookup: bool = False lookup_reference_field: str | None = None lookup_verify_field: str | None = None correction_days: int = Field(default=14, ge=1, le=365) class FormSettings(BaseModel): """Form-wide options.""" # People may save the form part-way ("Save draft") and finish it later. # A draft is visible only to whoever saved it and is left out of every # list, query and dashboard until it's submitted. allow_drafts: bool = False # Before submitting, people see a read-only summary of what they entered # and confirm it (or go back and edit). A Page can override this for the # form it embeds or opens — see PageBlockDef/PageAction.form_review. # Columns for fields that aren't inside any section/panel/tab (see # FormFieldDef.layout_columns). layout_columns: int | None = Field(default=None, ge=1, le=4) review_before_submit: bool = False # The summary's heading and a line of guidance above it; blank = defaults. review_title: str | None = Field(default=None, max_length=200) review_message: str | None = Field(default=None, max_length=1000) # A Print button: "off", on the review summary only ("review"), or on # the form too ("always"). It prints print_fields (blank = every field # with a value) as a read-only page headed print_title (blank = the # form's name). print_button: Literal["off", "review", "always"] = "off" print_fields: list[str] = [] print_title: str | None = Field(default=None, max_length=200) test_scenarios: list[FormTestScenario] = [] # language code ("hi", "ml", "ta-IN") -> that language's wording. translations: dict[Annotated[str, StringConstraints(pattern=r"^[a-z]{2,3}(-[A-Z]{2})?$")], FormTranslation] = {} # Form settings › Style — the look of the whole form, applied before # each field's own styles, and free CSS scoped to the form. styles: ElementStyles | None = None custom_css: CustomCss | None = None # Anyone with the form's public link can fill it in and submit it, without # signing in (app/routers/public_forms.py) — only fields and actions that # work for a signed-out visitor (app/crud/public_forms.py::public_problems). public_access: bool = False # Public submissions must pass reCAPTCHA (the platform's reCAPTCHA keys). public_captcha: bool = False # Shown once a public submission is saved; blank = a default thank-you. public_message: str | None = Field(default=None, max_length=1000) # A PDF of each submission, and an emailed receipt — see FormDocument / FormReceipt. document: FormDocument = FormDocument() receipt: FormReceipt = FormReceipt() schedule: FormSchedule = FormSchedule() review: FormReview = FormReview() class FormScenarioRun(BaseModel): """Run scenarios against a (possibly unsaved) design.""" field_definitions: list[FormFieldDef] scenarios: list[FormTestScenario] class FormScenarioResult(BaseModel): name: str passed: bool failures: list[str] errors: dict[str, str] values: dict class FormCreate(BaseModel): name: str field_definitions: list[FormFieldDef] = [] events: list[FormEventAction] = [] prefill: FormPrefillConfig | None = None settings: FormSettings = FormSettings() # "table": its submissions get a real table (app/crud/entity_storage.py) # straight away; until the first publish it follows the draft's fields. storage_tier: Literal["flexible", "table"] = "flexible" @model_validator(mode="after") def _check_prefill_mappings(self) -> "FormCreate": _check_prefill(self.prefill, self.field_definitions) _check_validation_refs(self.field_definitions) return self class FormUpdate(BaseModel): name: str | None = None field_definitions: list[FormFieldDef] | None = None events: list[FormEventAction] | None = None prefill: FormPrefillConfig | None = None settings: FormSettings | None = None # The updated_at last read — a save over a newer change is refused (409). expected_updated_at: datetime | None = None @model_validator(mode="after") def _check_prefill_mappings(self) -> "FormUpdate": _check_prefill(self.prefill, self.field_definitions) _check_validation_refs(self.field_definitions) return self class FormFieldsAdd(BaseModel): """Fields appended to an already-published Form (POST /forms/{id}/fields).""" field_definitions: list[FormFieldDef] = Field(min_length=1) @model_validator(mode="after") def _check_refs(self) -> "FormFieldsAdd": _check_validation_refs(self.field_definitions) return self class FormFieldAccessRules(BaseModel): """One field's access_rules (PUT /forms/{id}/fields/{name}/access-rules) — settable on a live form: permissions change who may see or edit a value, never the form's structure or its submissions.""" access_rules: list[FormFieldAccessRule] = [] class FormIndexedFields(BaseModel): """Which of the current version's fields are indexed (PUT /forms/{id}/indexed-fields) — settable on a live form, since an index changes nothing about the form itself.""" field_names: list[str] class FormRead(BaseModel): model_config = ConfigDict(from_attributes=True) form_id: uuid.UUID version: int tenant_id: uuid.UUID name: str field_definitions: list[dict] events: list[dict] prefill: FormPrefillConfig | None settings: FormSettings = FormSettings() status: PublishStatus is_current: bool created_by: uuid.UUID | None created_at: datetime updated_at: datetime # Not a Form column — filled in by the list and create routes. storage_tier: str = "flexible" class WorkflowStepFormCreate(BaseModel): step_key: str form_id: uuid.UUID form_version: int class WorkflowStepFormRead(BaseModel): model_config = ConfigDict(from_attributes=True) workflow_id: uuid.UUID workflow_version: int step_key: str form_id: uuid.UUID form_version: int # ---- Workflow instances & step history ------------------------------------------------------------- class WorkflowInstanceCreate(BaseModel): workflow_id: uuid.UUID initial_step: str vendor_id: uuid.UUID | None = None # Explicit deadline wins; omit it to auto-compute one from the Start # step's own sla_hours (if it declares one) — see _start_instance. sla_due_at: datetime | None = None # Only consulted for that auto-computation: when true, sla_hours is # spent only across Mon-Fri (a statutory "N business days to respond" # clock skips weekends entirely rather than counting them toward the # deadline). No holiday calendar — weekends only. sla_business_days: bool = False # Checked against the Start step's declared input_params before the # instance is created — see app/routers/workflow_instances.py. context: dict = {} # Set when this instance is being started as a child of another instance # (e.g. a Transfer/Information-Request branch) — usually set internally # via WorkflowInstanceAdvance.spawn rather than passed directly here, but # exposed for a direct/manual child start too. parent_instance_id: uuid.UUID | None = None class WorkflowInstanceRead(BaseModel): model_config = ConfigDict(from_attributes=True) id: uuid.UUID tenant_id: uuid.UUID vendor_id: uuid.UUID | None workflow_id: uuid.UUID workflow_version: int current_step: str | None status: InstanceStatus sla_due_at: datetime | None # sla_paused_at: set while the clock is currently paused (null = running). # sla_paused_seconds: total time already spent paused across every past # pause/resume cycle — added to sla_due_at (plus the current pause, if # any is still open) to get the *effective* deadline below, so pausing # never eats into the statutory window it was meant to protect. sla_paused_at: datetime | None sla_paused_seconds: int started_by: uuid.UUID | None started_at: datetime completed_at: datetime | None context: dict parent_instance_id: uuid.UUID | None # Where the instance is: one token per active branch (app/crud/workflow_engine.py). tokens: list[dict] = [] next_due_at: datetime | None = None suspended_at: datetime | None = None cancel_reason: str | None = None has_incident: bool = False has_escalation: bool = False @computed_field # type: ignore[prop-decorator] @property def effective_sla_due_at(self) -> datetime | None: if self.sla_due_at is None: return None paused = timedelta(seconds=self.sla_paused_seconds) if self.sla_paused_at is not None: paused += datetime.now(timezone.utc) - self.sla_paused_at return self.sla_due_at + paused @computed_field # type: ignore[prop-decorator] @property def is_overdue(self) -> bool: due = self.effective_sla_due_at return due is not None and self.status == "in_progress" and self.sla_paused_at is None and datetime.now(timezone.utc) > due class SpawnChildInstance(BaseModel): """One entry of WorkflowInstanceAdvance.spawn — same shape as starting an instance directly, since spawning a child IS starting a new instance, just with a parent link attached.""" workflow_id: uuid.UUID initial_step: str context: dict = {} class WorkflowInstanceAdvance(BaseModel): # Which branch moves, when there are several (default: the first one). token_id: str | None = None next_step: str action: str # e.g. approved, rejected, routed, escalated comment: str | None = None complete: bool = False # true if next_step is a terminal End step # New/updated context values the actor is providing at this transition — # validated against the leaving step's output_params, then merged into # the instance's context, then the entering step's input_params are # checked against that merged context. context: dict = {} # Child instances to start as part of this transition (e.g. a Transfer # decision's destination office(s)) — each becomes its own instance with # parent_instance_id set to this one. `complete=True` on an instance # that still has open (non-terminal) children is honored for the # transition itself but the instance is NOT actually marked completed # until every child finishes — see app/routers/workflow_instances.py. spawn: list[SpawnChildInstance] = [] class InstanceCancel(BaseModel): reason: str = Field(min_length=1, max_length=1000) class TokenAction(BaseModel): """An administrator's action on one branch of an instance.""" token_id: str = Field(min_length=1, max_length=40) comment: str | None = Field(default=None, max_length=2000) class TokenReassign(TokenAction): # The step now waits on these people instead of its approvers. user_ids: list[uuid.UUID] = [] role_ids: list[uuid.UUID] = [] class TokenMove(TokenAction): # The step to move this branch to (any step of the workflow). to: str = Field(min_length=1, max_length=100) class WorkflowDelegationCreate(BaseModel): # Whose tasks (default: your own; someone else's needs Workflows edit). user_id: uuid.UUID | None = None delegate_id: uuid.UUID starts_at: datetime ends_at: datetime workflow_id: uuid.UUID | None = None reason: str | None = Field(default=None, max_length=300) class WorkflowDelegationRead(BaseModel): model_config = ConfigDict(from_attributes=True) id: uuid.UUID user_id: uuid.UUID delegate_id: uuid.UUID starts_at: datetime ends_at: datetime workflow_id: uuid.UUID | None reason: str | None created_at: datetime user_name: str | None = None delegate_name: str | None = None workflow_name: str | None = None class StepHistoryRead(BaseModel): model_config = ConfigDict(from_attributes=True) id: uuid.UUID tenant_id: uuid.UUID instance_id: uuid.UUID step: str actor_id: uuid.UUID | None action: str comment: str | None occurred_at: datetime actor_name: str | None = None class FormSubmissionCreate(BaseModel): form_id: uuid.UUID form_version: int workflow_instance_id: uuid.UUID | None = None field_values: dict # Save part-way, to finish later (the form must allow drafts): required # fields and validation rules aren't checked yet, and no events fire. draft: bool = False # field name -> the token confirming its email/phone (require_verification). verifications: dict[str, str] = {} # Super Admin only (app/routers/platform_admin.py) — no regular route # updates a submission's field_values today, so this didn't exist before. class FormSubmissionUpdate(BaseModel): field_values: dict # For a draft: true submits it (full validation, then it counts like any # other submission); false/absent keeps it a draft. submit: bool = False # The updated_at last read — a save over a newer change is refused (409). expected_updated_at: datetime | None = None # field name -> the token confirming its email/phone (require_verification). verifications: dict[str, str] = {} class FormVerificationSend(BaseModel): form_id: uuid.UUID field: str value: str = Field(max_length=320) class FormVerificationSent(BaseModel): verification_id: uuid.UUID # Where the code went, partly hidden ("a•••@example.com", "•••• 2345"). sent_to: str class FormVerificationConfirm(BaseModel): form_id: uuid.UUID verification_id: uuid.UUID code: str = Field(max_length=12) class FormVerificationToken(BaseModel): # Sent back with the submission (FormSubmissionCreate.verifications). token: str # ---- Public forms (app/routers/public_forms.py) ------------------------------ class PublicFormRead(BaseModel): """A form open to the public, as a signed-out visitor sees it.""" tenant_name: str form_id: uuid.UUID version: int name: str field_definitions: list[dict] # Fields read-only for everyone (hidden ones aren't sent at all). field_access: dict[str, str] = {} events: list[dict] = [] settings: dict = {} # Set when the form asks for reCAPTCHA and the platform has its keys. recaptcha_site_key: str | None = None # "open", or why it isn't taking submissions: "scheduled" (not yet), # "paused", "closed", "archived" — with the line to show (form_schedule.py). availability: str = "open" availability_message: str | None = None opens_at: datetime | None = None closes_at: datetime | None = None class PublicFormOption(BaseModel): id: uuid.UUID field_values: dict class PublicVerificationSend(BaseModel): field: str value: str = Field(max_length=320) class PublicVerificationConfirm(BaseModel): verification_id: uuid.UUID code: str = Field(max_length=12) class PublicFormSubmit(BaseModel): field_values: dict verifications: dict[str, str] = {} captcha_token: str | None = Field(default=None, max_length=4000) # Submitting a draft saved with "continue later" — its link's token. resume_token: str | None = Field(default=None, max_length=100) class PublicDraftSave(BaseModel): field_values: dict # Saving again over the same draft — its link's token. resume_token: str | None = Field(default=None, max_length=100) class PublicDraftSaved(BaseModel): resume_token: str expires_at: datetime class PublicDraftRead(BaseModel): field_values: dict expires_at: datetime class PublicFormSubmitted(BaseModel): id: uuid.UUID message: str | None = None # label -> the reference number an auto-number field gave the submission. references: dict[str, str] = {} # When the form has a PDF document: a short-lived token to download it. document_token: str | None = None class FormSubmissionRead(BaseModel): model_config = ConfigDict(from_attributes=True) id: uuid.UUID tenant_id: uuid.UUID vendor_id: uuid.UUID | None form_id: uuid.UUID form_version: int workflow_instance_id: uuid.UUID | None submitted_by: uuid.UUID | None updated_by: uuid.UUID | None field_values: dict submitted_at: datetime created_at: datetime updated_at: datetime is_deleted: bool status: str = "submitted" # Officers' review (app/crud/application_review.py); None = submitted. review_status: str | None = None review_note: str | None = None reviewed_at: datetime | None = None correction_fields: list[str] | None = None correction_until: datetime | None = None # Filled in by the submissions list endpoint (app/crud/dataset_query.py:: # attach_user_names); None elsewhere. created_by_name is the submitter. created_by_name: str | None = None updated_by_name: str | None = None # ---- Application review (app/crud/application_review.py) ------------------------------------------- ReviewStatus = Literal["submitted", "under_review", "returned", "resubmitted", "approved", "rejected", "closed"] # What an officer can set (submitted / resubmitted come from the applicant). OfficerReviewStatus = Literal["under_review", "returned", "approved", "rejected", "closed"] class SubmissionReviewAction(BaseModel): status: OfficerReviewStatus # Required when returning or rejecting — the applicant sees it. note: str | None = Field(default=None, max_length=4000) # Returned only: the fields the applicant may change (blank = all). correction_fields: list[str] = [] expected_updated_at: datetime | None = None class SubmissionReviewRead(BaseModel): model_config = ConfigDict(from_attributes=True) id: uuid.UUID from_status: str | None to_status: str note: str | None actor_id: uuid.UUID | None actor_name: str | None = None occurred_at: datetime class SubmissionResubmit(BaseModel): field_values: dict verifications: dict[str, str] = {} class PublicStatusSend(BaseModel): reference: str = Field(min_length=1, max_length=100) class PublicStatusConfirm(BaseModel): reference: str = Field(min_length=1, max_length=100) verification_id: uuid.UUID code: str = Field(max_length=12) class PublicStatusToken(BaseModel): token: str class PublicStatusView(BaseModel): token: str = Field(max_length=2000) class PublicStatusEvent(BaseModel): status: str label: str note: str | None = None at: datetime class PublicStatusRead(BaseModel): """What an applicant sees of their application — no officer names.""" reference: str form_name: str status: str label: str note: str | None = None submitted_at: datetime history: list[PublicStatusEvent] = [] # Returned for correction and still in time: what they may change. can_correct: bool = False correction_until: datetime | None = None correction_fields: list[str] | None = None field_definitions: list[dict] = [] field_values: dict = {} class PublicStatusResubmit(BaseModel): token: str = Field(max_length=2000) field_values: dict verifications: dict[str, str] = {} class FormDuplicate(BaseModel): # Blank = "Copy of ". name: str | None = Field(default=None, max_length=200)