"""The app model — one document describing an app (or the whole tenant): entities the data: datasources, forms (a form's submissions are its entity's records) and data queries (external, read-only), all in one field vocabulary, with their storage tier; relations every link between entities (a reference field, a choice backed by a datasource or a query), and whether the database enforces it; views what people see — forms, pages, dashboards, reports — each with the entities it shows or writes; logic what runs — tasks and workflows — with the entities they touch; security roles, record access policies and field permissions; issues references to things that no longer exist. Every designer still edits its own part; this is the one place they come together — what the App Model screen draws, what an app's version is computed from, and what promotion and AI editing build on next. Which entities a view or logic item uses is found by walking its definition for entity ids (in any key, and in "master:"-style data sources), so new block or node types are covered without changes here. """ import hashlib import json import re import uuid from datetime import datetime, timezone from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.crud import record_sql from app.crud.entity_storage import tables_by_source from app.models.access import RecordAccessPolicy from app.models.app_bundle import App from app.models.automation import Task from app.models.data_query import DataQuery from app.models.identity import Role from app.models.master import Master from app.models.navigation import Dashboard, Page, Widget from app.models.report_template import ReportTemplate from app.models.workflow import Form, Workflow, WorkflowStepForm # ---- one field vocabulary ---------------------------------------------------------------------------- _DATASOURCE_TYPES = { "text": "text", "varchar": "text", "char": "text", "inet": "text", "number": "number", "numeric": "number", "real": "number", "double_precision": "number", "smallint": "integer", "integer": "integer", "bigint": "integer", "boolean": "boolean", "date": "date", "time": "time", "timestamp": "datetime", "timestamptz": "datetime", "interval": "duration", "uuid": "id", "json": "json", "jsonb": "json", "single_select": "choice", "reference": "reference", "formula": "formula", "file": "file", "user": "user", "geolocation": "location", } _FORM_TYPES = { "text": "text", "password": "text", "email": "text", "url": "text", "color_picker": "text", "label": "text", "hidden": "text", "textarea": "long_text", "rich_text": "long_text", "markdown": "long_text", "html": "long_text", "code": "long_text", "number": "number", "currency": "number", "percentage": "number", "slider": "number", "rating": "number", "checkbox": "boolean", "date": "date", "time": "time", "datetime": "datetime", "daterange": "date_range", "duration": "duration", "radio": "choice", "dropdown": "choice", "autocomplete": "choice", "combobox": "choice", "lookup": "choice", "multiselect": "multi_choice", "tags": "multi_choice", "treeselect": "multi_choice", "user": "user", "file_upload": "file", "image_upload": "file", "attachment": "file", "multi_attachment": "file", "signature": "file", "json": "json", "formula": "formula", "dynamic_grid": "table", "subform": "table", "geolocation": "location", "phone": "text", "barcode": "text", "autonumber": "text", "address": "json", "toggle": "boolean", "consent": "boolean", "matrix": "json", "ranking": "multi_choice", "camera": "file", } # A form's layout and buttons belong to the view, not the data. _FORM_VIEW_ONLY = {"section", "panel", "tabs", "accordion", "wizard", "row", "column", "group_end", "button"} _UUID = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", re.IGNORECASE) _SOURCE_REF = re.compile(r"^(master|form|query|dataset):([0-9a-f-]{36})$", re.IGNORECASE) # Keys whose value names an entity — checked for references that no longer resolve. _ENTITY_KEYS = {"master_id", "form_id", "reference_master_id", "reference_form_id", "options_query_id", "query_id", "target_form_id", "embedded_form_id", "label_source_id"} def _tier(table: Any) -> str: return "flexible" if table is None else "building" if table.state == "building" else "table" def _field(name: str, label: str, type_: str, source_type: str, *, required: bool = False, **extra: Any) -> dict: return {"name": name, "label": label, "type": type_, "source_type": source_type, "required": required, **{k: v for k, v in extra.items() if v not in (None, [], False)}} def _datasource_fields(field_schema: list[dict]) -> list[dict]: out = [] for f in field_schema: t = f.get("type", "") many = bool(f.get("multi")) target = f.get("reference_master_id") or f.get("reference_form_id") out.append( _field( f["name"], f.get("alias") or f["name"], _DATASOURCE_TYPES.get(t, "text"), t, required=bool(f.get("required")), options=f.get("options"), reference={"entity_id": str(target), "many": many, "display_field": f.get("reference_display_field")} if t == "reference" and target else None, indexed=bool(f.get("indexed")), formula=f.get("formula_expression"), ) ) return out def _form_fields(defs: list[dict]) -> list[dict]: out = [] for f in defs: t = f.get("type", "") if t in _FORM_VIEW_ONLY or not f.get("name"): continue target = f.get("master_id") or f.get("options_query_id") or f.get("embedded_form_id") or (str(f["link_source"]).partition(":")[2] if f.get("link_source") else None) unified = _FORM_TYPES.get(t, "text") out.append( _field( f["name"], f.get("label") or f["name"], "reference" if target else unified, t, required=bool(f.get("required")), options=f.get("options") if not target else None, reference={"entity_id": str(target), "many": unified == "multi_choice", "display_field": f.get("master_display_field") or f.get("options_label_field") or f.get("link_display_field")} if target else None, indexed=bool(f.get("indexed")), formula=f.get("formula_expression"), restricted=bool(f.get("access_rules")), fields=_form_fields(f.get("subform_fields") or []) or None, ) ) return out # ---- walking definitions for the entities they use ------------------------------------------------------ def _walk(value: Any, known: set[str], used: set[str], missing: set[str], key: str | None = None) -> None: if isinstance(value, dict): for k, v in value.items(): _walk(v, known, used, missing, k) elif isinstance(value, list): for v in value: _walk(v, known, used, missing, key) elif isinstance(value, str): m = _SOURCE_REF.match(value) ident = m.group(2).lower() if m else value.lower() if _UUID.match(value) else None if ident is None: return if ident in known: used.add(ident) elif (m and m.group(1).lower() != "dataset") or key in _ENTITY_KEYS: missing.add(value) def _uses(definition: Any, known: set[str]) -> tuple[list[str], list[str]]: used: set[str] = set() missing: set[str] = set() _walk(definition, known, used, missing) return sorted(used), sorted(missing) # ---- building ---------------------------------------------------------------------------------------------- def _count_fields(count: tuple[int, bool] | None) -> dict: total, capped = count or (0, False) return {"record_count": total, "record_count_capped": capped} async def build_model(db: AsyncSession, tenant_id: uuid.UUID, app: App | None = None) -> dict: masters = (await db.execute(select(Master).where(Master.tenant_id == tenant_id).order_by(Master.name))).scalars().all() forms = (await db.execute(select(Form).where(Form.tenant_id == tenant_id, Form.is_current.is_(True)).order_by(Form.name))).scalars().all() queries = (await db.execute(select(DataQuery).where(DataQuery.tenant_id == tenant_id).order_by(DataQuery.name))).scalars().all() pages = (await db.execute(select(Page).where(Page.tenant_id == tenant_id).order_by(Page.name))).scalars().all() dashboards = (await db.execute(select(Dashboard).where(Dashboard.tenant_id == tenant_id).order_by(Dashboard.name))).scalars().all() widgets = (await db.execute(select(Widget).where(Widget.tenant_id == tenant_id))).scalars().all() reports = ( await db.execute(select(ReportTemplate).where(ReportTemplate.tenant_id == tenant_id, ReportTemplate.vendor_id.is_(None)).order_by(ReportTemplate.name)) ).scalars().all() tasks = (await db.execute(select(Task).where(Task.tenant_id == tenant_id).order_by(Task.name))).scalars().all() workflows = (await db.execute(select(Workflow).where(Workflow.tenant_id == tenant_id, Workflow.is_current.is_(True)).order_by(Workflow.name))).scalars().all() step_forms = ( (await db.execute(select(WorkflowStepForm).where(WorkflowStepForm.workflow_id.in_([w.workflow_id for w in workflows])))).scalars().all() if workflows else [] ) roles = (await db.execute(select(Role).where(Role.tenant_id == tenant_id).order_by(Role.name))).scalars().all() policies = (await db.execute(select(RecordAccessPolicy).where(RecordAccessPolicy.tenant_id == tenant_id))).scalars().all() tables = await tables_by_source(db, tenant_id) master_counts = await record_sql.capped_source_counts(db, "master", tenant_id, [m.id for m in masters]) form_counts = await record_sql.capped_source_counts(db, "form", tenant_id, [f.form_id for f in forms]) # ---- entities entities: list[dict] = [] for m in masters: table = tables.get(str(m.id)) enforced = {f for f, c in (table.columns.items() if table else []) if c.get("ref_table")} entities.append({ "id": str(m.id), "kind": "datasource", "name": m.name, "fields": _datasource_fields(m.field_schema), "storage": {"tier": _tier(table), "table": f"entity_data.{table.table_name}" if table else None, "enforced_links": sorted(enforced)}, **_count_fields(master_counts.get(m.id)), }) for f in forms: table = tables.get(str(f.form_id)) entities.append({ "id": str(f.form_id), "kind": "form", "name": f.name, "fields": _form_fields(f.field_definitions), "storage": {"tier": _tier(table), "table": f"entity_data.{table.table_name}" if table else None, "enforced_links": []}, **_count_fields(form_counts.get(f.form_id)), }) for q in queries: entities.append({ "id": str(q.id), "kind": "query", "name": q.name, "fields": [_field(c, c, "text", "column") for c in (q.columns or [])], "storage": {"tier": "external", "table": None, "enforced_links": []}, "record_count": None, "record_count_capped": False, }) known = {e["id"] for e in entities} by_id = {e["id"]: e for e in entities} # ---- relations relations: list[dict] = [] for e in entities: enforced = set(e["storage"]["enforced_links"]) for fld in e["fields"]: ref = fld.get("reference") if ref: relations.append({ "id": f"{e['id']}.{fld['name']}", "from": e["id"], "field": fld["name"], "to": ref["entity_id"], "many": ref["many"], "enforced": fld["name"] in enforced, "resolved": ref["entity_id"] in known, }) # ---- views and logic issues: list[dict] = [] def item(kind: str, id_: str, name: str, definition: Any, status: str | None = None, primary: str | None = None) -> dict: used, missing = _uses(definition, known) if primary and primary not in used: used = [primary, *used] for ref in missing: issues.append({"severity": "warning", "item_kind": kind, "item_id": id_, "item_name": name, "message": f"{name} refers to something that no longer exists ({ref})."}) return {"id": id_, "kind": kind, "name": name, "status": status, "entity_ids": used} widgets_by_dashboard: dict[str, list] = {} for w in widgets: widgets_by_dashboard.setdefault(str(w.dashboard_id), []).append({"data_source": w.data_source, "config": w.config}) forms_by_workflow: dict[str, list[str]] = {} for sf in step_forms: forms_by_workflow.setdefault(str(sf.workflow_id), []).append(str(sf.form_id)) views = [ *(item("form", str(f.form_id), f.name, [f.field_definitions, f.events, f.prefill, f.settings], f.status, primary=str(f.form_id)) for f in forms), *(item("page", str(p.id), p.name, [p.blocks, p.events, p.variables], p.status) for p in pages), *(item("dashboard", str(d.id), d.name, [d.parameters, widgets_by_dashboard.get(str(d.id), [])], d.status) for d in dashboards), *(item("report", str(r.id), r.name, [r.sections, r.parameters]) for r in reports), ] logic = [ *(item("task", str(t.id), t.name, t.nodes, t.status) for t in tasks), *(item("workflow", str(w.workflow_id), w.name, [w.step_definitions, forms_by_workflow.get(str(w.workflow_id), [])], w.status) for w in workflows), ] for rel in relations: if not rel["resolved"]: src = by_id[rel["from"]] issues.append({"severity": "warning", "item_kind": src["kind"], "item_id": src["id"], "item_name": src["name"], "message": f"{src['name']}.{rel['field']} links to something that no longer exists."}) # ---- security policy_by_entity = {str(p.source_id): p for p in policies} security = { "roles": [{"id": str(r.id), "name": r.name, "scope": r.scope, "system": r.is_system_role} for r in roles], "policies": [ {"entity_id": str(p.source_id), "enabled": p.enabled, "rule_count": len(p.rules or [])} for p in policies if str(p.source_id) in known ], } for e in entities: p = policy_by_entity.get(e["id"]) e["access"] = { "policy": bool(p and p.enabled), "restricted_fields": [f["name"] for f in e["fields"] if f.get("restricted")], } # ---- which app scope: dict = {"kind": "tenant", "id": None, "name": "All apps"} if app is not None: members = {(i["type"], str(i["id"])) for i in (app.items or [])} member_ids = {i for _, i in members} views = [v for v in views if (v["kind"], v["id"]) in members or (v["kind"] == "form" and ("form", v["id"]) in members)] logic = [g for g in logic if (g["kind"], g["id"]) in members] used = {eid for x in (*views, *logic) for eid in x["entity_ids"]} | {i for t, i in members if t in ("master", "form")} # Entities a member links to come along too, marked as outside the app. frontier = set(used) while frontier: nxt = {r["to"] for r in relations if r["from"] in frontier and r["resolved"]} - used used |= nxt frontier = nxt entities = [e for e in entities if e["id"] in used] relations = [r for r in relations if r["from"] in used] issues = [i for i in issues if i["item_id"] in used or i["item_id"] in {x["id"] for x in (*views, *logic)}] security["policies"] = [p for p in security["policies"] if p["entity_id"] in used] for e in entities: e["in_app"] = e["id"] in member_ids scope = {"kind": "app", "id": str(app.id), "name": app.name} else: for e in entities: e["in_app"] = True # ---- who uses each entity for e in entities: e["used_by"] = [{"kind": x["kind"], "id": x["id"], "name": x["name"]} for x in (*views, *logic) if e["id"] in x["entity_ids"]] body = {"scope": scope, "entities": entities, "relations": relations, "views": views, "logic": logic, "security": security, "issues": issues} digest = hashlib.sha256(json.dumps(body, sort_keys=True, default=str).encode()).hexdigest() return { **body, "version": digest[:12], "generated_at": datetime.now(timezone.utc).isoformat(), "summary": { "entities": len(entities), "relations": len(relations), "views": len(views), "logic": len(logic), "tables": sum(1 for e in entities if e["storage"]["tier"] == "table"), "issues": len(issues), }, }