import uuid from datetime import datetime from typing import Literal from pydantic import BaseModel, ConfigDict, Field, model_validator from app.schemas.validation_rules import FormValidationRule # FR-MST-002's "at minimum" set is text/number/date/boolean/single_select; # this is the fuller PostgreSQL-scalar-type picker described in # app/crud/validation.py's module docstring. Kept as flat string literals # (not int/float/etc. Python types) since these describe how a value in # the master_records.field_values JSONB column should be interpreted and # validated — there's no real per-column Postgres type behind any of this. MasterFieldType = Literal[ "text", "varchar", "char", "number", "smallint", "integer", "bigint", "numeric", "real", "double_precision", "boolean", "date", "time", "timestamp", "timestamptz", "interval", "uuid", "json", "jsonb", "inet", "single_select", "reference", # dropdown live-linked to another Master's records (e.g. State.country -> Country) "formula", # server-computed from other fields on the same record — see app/crud/formula.py "file", # uploaded via POST /uploads, stored as that upload's id (see app/routers/uploads.py) "user", # dropdown live-linked to this tenant's system Users — see app/crud/user_picker.py "geolocation", # a map-picked point, stored as "lat,lng" text — see app/crud/validation.py "autonumber", # given the next number once, on create, in autonumber_format — see app/crud/field_constraints.py ] UserPickerType = Literal["internal", "portal", "any"] # How a field is edited (Datasources › field › Control) — the first control # listed for a type is its default. Each keeps the type's stored value, so a # control can change at any time without touching the data. Mirrors # frontend/src/lib/masterControls.ts. _TEXT_CONTROLS = ["text", "textarea", "email", "url", "phone", "password", "rich_text", "color"] _NUMBER_CONTROLS = ["number", "currency", "percentage", "slider", "rating"] MASTER_FIELD_CONTROLS: dict[str, list[str]] = { "text": ["textarea", *[c for c in _TEXT_CONTROLS if c != "textarea"]], "varchar": ["text", "email", "url", "phone", "password", "color"], "char": ["text", "email", "url", "phone", "password", "color"], **{t: _NUMBER_CONTROLS for t in ("number", "smallint", "integer", "bigint", "numeric", "real", "double_precision")}, "boolean": ["checkbox", "toggle", "yes_no"], "single_select": ["dropdown", "radio", "autocomplete"], "user": ["dropdown", "autocomplete"], } # "reference": one record — dropdown / autocomplete / radio; several — checkboxes / multiselect. _REFERENCE_CONTROLS = {False: ["dropdown", "autocomplete", "radio"], True: ["checkboxes", "multiselect"]} def controls_for(field_type: str, multi: bool = False) -> list[str]: """The controls a field of this type can be edited with (default first).""" if field_type == "reference": return _REFERENCE_CONTROLS[bool(multi)] return MASTER_FIELD_CONTROLS.get(field_type, []) class MasterFieldDef(BaseModel): name: str # Display name shown to people wherever this column appears (tables, # cards, forms, reports) instead of the raw `name`, which stays the stable # key stored in records and referenced by filters/formulas. Blank = show # `name`. alias: str | None = None type: MasterFieldType required: bool = False options: list[str] | None = None # single_select only max_length: int | None = None # varchar/char only reference_master_id: uuid.UUID | None = None # reference only — which Master this field links to # reference only — alternatively link to a Form's submissions instead of a # Master's records (exactly one of reference_master_id / reference_form_id). # The stored value is then a submission id, and reference_display_field # names one of that Form's fields. reference_form_id: uuid.UUID | None = None reference_display_field: str | None = None # reference only — which of that Master's fields to show as the option label multi: bool = False # reference only — select one linked record (default) or several formula_expression: str | None = None # formula only — e.g. "{quantity} * {unit_price}" accept: str | None = None # file only — MIME accept filter, e.g. "image/*" 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 # reference only — cascading/dependent filtering: cascade_parent_field # names another "reference" field on this SAME field_schema whose # currently-selected value narrows this field's options; cascade_match_field # names the field on reference_master_id's own schema whose value must # equal the parent's selection. Client-side convenience filter only (not # enforced by validate_field_values) — see app/schemas/master.py's # _check_cascade_fields for the cross-field structural check. cascade_parent_field: str | None = None cascade_match_field: str | None = None # 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: the source gets a trigram # index over its records, if any of its fields asks (field_indexes.py). searchable: bool = False # How the field is edited — one of controls_for(type, multi); blank = the # type's default. Display only: the stored value is the type's either way. control: str | None = None # Checked on every write (app/crud/validation.py) and in the record # editor — the same rules a Form field takes (app/core/form_validation.py). validation_rules: list[FormValidationRule] = [] # Shown instead of the default "… is required" when it's blank. required_message: str | None = Field(default=None, max_length=300) # No two records may hold the same value (text without case or spaces). unique: bool = False # Filled in on create when left blank: a value, or {{today}}, {{now}}, # {{current_user.id}} / .name / .email (app/crud/field_constraints.py). default_value: str | None = Field(default=None, max_length=500) # autonumber only — e.g. "APP-{YYYY}-{SEQ:5}": {SEQ} the running number # ({SEQ:5} zero-padded to 5), {YYYY} {YY} {MM} {DD} the date it's given. autonumber_format: str | None = Field(default=None, max_length=100) # autonumber only — start again at 1 each year / month. autonumber_reset: Literal["never", "yearly", "monthly"] = "never" @model_validator(mode="after") def _check_type_specific_fields(self) -> "MasterFieldDef": self.alias = (self.alias or "").strip() or None 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}': an auto-number's format needs {{SEQ}} — the running number") if self.required or self.default_value: raise ValueError(f"field '{self.name}': an auto-number is given out — it can't be required or have a default") elif self.autonumber_format is not None: raise ValueError("autonumber_format is only meaningful for type 'autonumber'") if self.type == "formula" and (self.default_value or self.unique): raise ValueError(f"field '{self.name}': a formula is computed — it can't have a default or be unique") if self.unique and self.type in ("json", "jsonb", "file", "geolocation") or self.unique and self.multi: raise ValueError(f"field '{self.name}': this kind of field can't be unique") if self.type == "single_select" and not self.options: raise ValueError("single_select fields need at least one option") if self.options is not None and self.type != "single_select": raise ValueError("options is only meaningful for single_select fields") if self.max_length is not None and self.type not in ("varchar", "char"): raise ValueError("max_length is only meaningful for varchar/char fields") if self.type == "reference": if (self.reference_master_id is None) == (self.reference_form_id is None): raise ValueError("reference fields link to exactly one target: a datasource (reference_master_id) or a form (reference_form_id)") if not self.reference_display_field: raise ValueError("reference fields need reference_display_field") else: if self.reference_master_id is not None or self.reference_form_id is not None or self.reference_display_field is not None: raise ValueError("reference_master_id/reference_form_id/reference_display_field are only meaningful for type 'reference'") if self.multi: raise ValueError("multi is only meaningful for type 'reference'") if self.type == "formula": if not self.formula_expression: raise ValueError("formula fields need a formula_expression") from app.core.expr import expression_error error = expression_error(self.formula_expression) if error: raise ValueError(f"field '{self.name}': formula — {error}") if self.required: raise ValueError("formula fields are server-computed and cannot be required") elif self.formula_expression is not None: raise ValueError("formula_expression is only meaningful for type 'formula'") if self.accept is not None and self.type != "file": raise ValueError("accept is only meaningful for type 'file'") 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'") if self.control is not None: allowed = controls_for(self.type, self.multi) if self.control not in allowed: raise ValueError( f"field '{self.name}': control '{self.control}' doesn't fit a {self.type} field" + (f" — choose one of {', '.join(allowed)}" if allowed else " — it has no other controls") ) if self.type == "formula" and self.validation_rules: raise ValueError("formula fields are server-computed and take no validation rules") if any(r.type in ("unique", "exists_in") for r in self.validation_rules): raise ValueError(f"field '{self.name}': use the field's Unique setting or a reference field — 'unique'/'exists_in' rules are for form fields") if self.type == "reference": if bool(self.cascade_parent_field) != bool(self.cascade_match_field): raise ValueError("cascade_parent_field and cascade_match_field must be set together") elif self.cascade_parent_field is not None or self.cascade_match_field is not None: raise ValueError("cascade_parent_field/cascade_match_field are only meaningful for type 'reference'") return self def _check_scope_field(field_schema: list[MasterFieldDef] | None, scope_field: str | None) -> None: """scope_field (see app/crud/scope.py) must name an existing `reference` field on this same field_schema. Only enforced when both are present in the same payload — an update that changes scope_field without resending field_schema (or vice versa) is trusted, since the Master schema editor always resubmits the full field_schema together with any scope_field change.""" if not scope_field or field_schema is None: return match = next((f for f in field_schema if f.name == scope_field), None) if match is None: raise ValueError(f"scope_field '{scope_field}' does not name a field on this Master") if match.type != "reference": raise ValueError(f"scope_field '{scope_field}' must be a 'reference' field") if match.reference_form_id is not None: raise ValueError(f"scope_field '{scope_field}' must link to a datasource — row-level scope can't be based on a form") def _check_cascade_fields(field_schema: list[MasterFieldDef] | None) -> None: """Each field's cascade_parent_field (see MasterFieldDef above) must name a DIFFERENT "reference" field on this same field_schema. Only enforced when field_schema is present in the payload — same trust posture as _check_scope_field.""" if field_schema is None: return by_name = {f.name: f for f in field_schema} for field in field_schema: if not field.cascade_parent_field: continue if field.cascade_parent_field == field.name: raise ValueError(f"'{field.name}' cannot cascade from itself") parent = by_name.get(field.cascade_parent_field) if parent is None: raise ValueError(f"'{field.name}' cascades from '{field.cascade_parent_field}', which isn't a field on this Master") if parent.type != "reference": raise ValueError(f"'{field.name}' cascades from '{field.cascade_parent_field}', which must be a 'reference' field") class MasterCreate(BaseModel): name: str field_schema: list[MasterFieldDef] = [] scope_field: str | None = None # "table": also create its real table (app/crud/entity_storage.py) now, # in the same transaction — so the datasource never exists without it. storage_tier: Literal["flexible", "table"] = "flexible" @model_validator(mode="after") def _check_scope(self) -> "MasterCreate": _check_scope_field(self.field_schema, self.scope_field) _check_cascade_fields(self.field_schema) return self class MasterUpdate(BaseModel): name: str | None = None field_schema: list[MasterFieldDef] | None = None scope_field: str | 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_scope(self) -> "MasterUpdate": _check_scope_field(self.field_schema, self.scope_field) _check_cascade_fields(self.field_schema) return self class MasterRead(BaseModel): model_config = ConfigDict(from_attributes=True) id: uuid.UUID tenant_id: uuid.UUID name: str field_schema: list[dict] scope_field: str | None created_by: uuid.UUID | None created_at: datetime updated_at: datetime record_count: int = 0 # not a Master column — computed by the router, see masters.py's _to_read record_count_capped: bool = False # there are more than record_count (record_sql.COUNT_CAP) # "flexible" (JSONB rows) or "table" (a real table too) — app/crud/entity_storage.py. storage_tier: str = "flexible" class StorageColumn(BaseModel): field: str label: str column: str sql_type: str required: bool options: list[str] | None = None max_length: int | None = None is_link: bool = False link_table: str | None = None link_enforced: bool = False indexed: bool = False class StorageIssue(BaseModel): field: str label: str problem: str # type | required | options | length | link message: str count: int record_ids: list[str] samples: list = [] class StorageMigration(BaseModel): id: str action: str # promote | migrate | demote summary: str statements: list[str] applied_by: str | None applied_at: str class StorageInfo(BaseModel): tier: str # flexible | building | table table: str | None columns: list[StorageColumn] row_count: int | None # While building: {copied, total} records. progress: dict | None = None history: list[StorageMigration] class StoragePreview(BaseModel): table: str columns: list[StorageColumn] issues: list[StorageIssue] class StorageChange(BaseModel): tier: Literal["flexible", "table"] class MasterRecordCreate(BaseModel): field_values: dict class MasterRecordUpdate(BaseModel): field_values: dict # The updated_at last read — a save over a newer change is refused (409). expected_updated_at: datetime | None = None class MasterRecordRead(BaseModel): model_config = ConfigDict(from_attributes=True) id: uuid.UUID tenant_id: uuid.UUID master_id: uuid.UUID field_values: dict created_by: uuid.UUID | None updated_by: uuid.UUID | None created_at: datetime updated_at: datetime is_deleted: bool # Filled in by the record list endpoint (app/crud/dataset_query.py:: # attach_user_names); None elsewhere. created_by_name: str | None = None updated_by_name: str | None = None class ScopeAssignmentInput(BaseModel): """One entry of a user's scope-assignment set — see app/crud/scope.py. `master_id` is the *scope-defining* Master (e.g. "Offices"), `record_id` one of its records (e.g. one specific Office).""" master_id: uuid.UUID record_id: uuid.UUID class ScopeAssignmentRead(BaseModel): model_config = ConfigDict(from_attributes=True) id: uuid.UUID user_id: uuid.UUID master_id: uuid.UUID record_id: uuid.UUID class BulkImportRowError(BaseModel): row: int errors: list[str] class BulkImportResult(BaseModel): total_rows: int valid_rows: int committed: bool row_errors: list[BulkImportRowError] class RecordImportStarted(BaseModel): job_id: uuid.UUID class RecordImportStatus(BaseModel): status: str # queued | running | done | failed filename: str done_rows: int imported: int failed: int errors: list[BulkImportRowError] message: str | None = None class ImportColumnsResult(BaseModel): columns: list[str]