""" App export/import — the actual cross-tenant transfer mechanism behind App Management (see routers/apps.py). An App (models/app_bundle.py) only stores *which* existing items belong to it; capture_bundle() reads their current content into a portable JSON payload, and import_bundle() replays that payload as a fresh set of creates in the caller's own tenant (i.e. "release to QA" = log into the QA tenant and import the file exported from wherever). Every cross-entity id in this codebase is already a *soft* reference (no DB FK between e.g. Menu.target_id and forms.form_id — every consuming UI already falls back to "Deleted X" when one doesn't resolve). Import exploits that directly: Pass 1 allocates a fresh id for every bundled item; Pass 2 re-inserts each item with its reference fields rewritten via that map where the target is also in the bundle, and left untouched (old id, gracefully "not found" in the new tenant) where it isn't. The one exception is `connector_id`, which — unlike the others — IS a real DB foreign key (widgets.connector_id -> connectors.id) and also holds credentials that must never leave their own environment, so it is always cleared to null rather than remapped or left dangling. Imported Forms/Workflows/Tasks/Pages/Dashboards are created via the exact same construction each already-existing create_* router endpoint uses, so they land as ordinary fresh drafts (version 1, is_current=True, status="draft") — nothing goes live in a new environment without someone deliberately publishing it there. Promotion: importing an app again from the same source environment updates the items its earlier import created (app_import_links) instead of adding duplicates — see import_bundle. Exports carry each Form's and Page's published version. """ import re import uuid from sqlalchemy import delete as sa_delete from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.html_sanitize import sanitize_document from app.models.access import PageVersion from app.models.app_bundle import AppImportLink from app.models.automation import Task from app.models.identity import Role from app.models.master import Master from app.models.navigation import Dashboard, Menu, MenuRole, Page, Widget from app.models.workflow import Form, Workflow, WorkflowStepForm from app.schemas.app_bundle import AppImportResult # ---- capture (export) ----------------------------------------------------------- _TYPES = ("master", "form", "workflow", "task", "page", "dashboard", "menu") async def capture_bundle(db: AsyncSession, tenant_id: uuid.UUID, app_name: str, description: str | None, items: list[dict]) -> dict: ids_by_type: dict[str, list[uuid.UUID]] = {t: [] for t in _TYPES} for item in items: # `items` comes straight off the App row's JSONB column, so ids are # plain strings there (JSON has no UUID type) — normalize once here. ids_by_type.setdefault(item["type"], []).append(uuid.UUID(str(item["id"]))) missing: list[str] = [] out: dict[str, list[dict]] = {t: [] for t in _TYPES} if ids_by_type["master"]: rows = (await db.execute(select(Master).where(Master.tenant_id == tenant_id, Master.id.in_(ids_by_type["master"])))).scalars().all() found = {r.id for r in rows} missing += [f"master {i}" for i in ids_by_type["master"] if i not in found] out["master"] = [{"id": str(r.id), "name": r.name, "field_schema": r.field_schema} for r in rows] if ids_by_type["form"]: # The live (latest published) version — what's tested and released # here — not a draft someone is still editing. from app.crud.form_versions import live_form rows = [r for r in [await live_form(db, tenant_id, i) for i in ids_by_type["form"]] if r is not None] found = {r.form_id for r in rows} missing += [f"form {i}" for i in ids_by_type["form"] if i not in found] out["form"] = [ { "id": str(r.form_id), "name": r.name, "field_definitions": r.field_definitions, "events": r.events, "prefill": r.prefill, "status": r.status, "version": r.version, } for r in rows ] if ids_by_type["workflow"]: rows = ( await db.execute( select(Workflow).where( Workflow.tenant_id == tenant_id, Workflow.workflow_id.in_(ids_by_type["workflow"]), Workflow.is_current.is_(True) ) ) ).scalars().all() found = {r.workflow_id for r in rows} missing += [f"workflow {i}" for i in ids_by_type["workflow"] if i not in found] for r in rows: step_forms = ( await db.execute( select(WorkflowStepForm).where( WorkflowStepForm.workflow_id == r.workflow_id, WorkflowStepForm.workflow_version == r.version ) ) ).scalars().all() out["workflow"].append( { "id": str(r.workflow_id), "name": r.name, "step_definitions": r.step_definitions, "settings": r.settings or {}, "status": r.status, "step_forms": [ {"step_key": sf.step_key, "form_id": str(sf.form_id), "form_version": sf.form_version} for sf in step_forms ], } ) if ids_by_type["task"]: rows = (await db.execute(select(Task).where(Task.tenant_id == tenant_id, Task.id.in_(ids_by_type["task"])))).scalars().all() found = {r.id for r in rows} missing += [f"task {i}" for i in ids_by_type["task"] if i not in found] out["task"] = [{"id": str(r.id), "name": r.name, "nodes": r.nodes, "variables": r.variables or [], "status": r.status} for r in rows] if ids_by_type["page"]: rows = (await db.execute(select(Page).where(Page.tenant_id == tenant_id, Page.id.in_(ids_by_type["page"])))).scalars().all() found = {r.id for r in rows} missing += [f"page {i}" for i in ids_by_type["page"] if i not in found] # The published version where there is one (else the working copy), # with everything that makes up the page. published = {} wanted = [(r.id, r.published_version) for r in rows if r.published_version is not None] if wanted: versions = await db.execute(select(PageVersion).where(PageVersion.page_id.in_([pid for pid, _ in wanted]))) published = {(v.page_id, v.version): v for v in versions.scalars()} for r in rows: src = published.get((r.id, r.published_version), r) out["page"].append( { "id": str(r.id), "name": src.name, "blocks": src.blocks, "parameters": src.parameters, "events": src.events, "canvas_width": src.canvas_width, "fit_mode": src.fit_mode, "status": r.status, "content_mode": src.content_mode, "html_document": src.html_document, "seo": src.seo or {}, } ) if ids_by_type["dashboard"]: rows = ( await db.execute(select(Dashboard).where(Dashboard.tenant_id == tenant_id, Dashboard.id.in_(ids_by_type["dashboard"]))) ).scalars().all() found = {r.id for r in rows} missing += [f"dashboard {i}" for i in ids_by_type["dashboard"] if i not in found] for r in rows: widgets = (await db.execute(select(Widget).where(Widget.dashboard_id == r.id).order_by(Widget.sort_order))).scalars().all() out["dashboard"].append( { "id": str(r.id), "name": r.name, "status": r.status, "parameters": r.parameters, "widgets": [ { "type": w.type, "data_source": w.data_source, "size": w.size, "color": w.color, "refresh_interval": w.refresh_interval, "connector_id": str(w.connector_id) if w.connector_id else None, "sort_order": w.sort_order, "config": w.config, } for w in widgets ], } ) if ids_by_type["menu"]: rows = (await db.execute(select(Menu).where(Menu.tenant_id == tenant_id, Menu.id.in_(ids_by_type["menu"])))).scalars().all() found = {r.id for r in rows} missing += [f"menu {i}" for i in ids_by_type["menu"] if i not in found] out["menu"] = [ { "id": str(r.id), "label": r.label, "icon": r.icon, "link_type": r.link_type, "linked_screen": r.linked_screen, "target_type": r.target_type, "target_id": str(r.target_id) if r.target_id else None, "group_label": r.group_label, "feature_name": r.feature_name, "feature_status": r.feature_status, "external_link_url": r.external_link_url, "sort_order": r.sort_order, "status": r.status, "audience": r.audience, } for r in rows ] return { "app_name": app_name, "description": description, "source_tenant_id": str(tenant_id), "missing_items": missing, "items": out, } # ---- import (release) ------------------------------------------------------------ _BUNDLE_TARGET_TYPES = ("master", "form", "page", "dashboard") # what Menu/PageAction/Form-event target_type can be def _new_id_map(items: dict[str, list[dict]]) -> dict[str, str]: """old_id (str) -> new_id (str), across every type at once — a bundled Form and a bundled Master never collide on id, so one flat map is fine.""" id_map: dict[str, str] = {} for type_items in items.values(): for item in type_items: id_map[item["id"]] = str(uuid.uuid4()) return id_map def _remap(old_id: str | None, id_map: dict[str, str]) -> str | None: if old_id is None: return None return id_map.get(old_id, old_id) def _remap_data_source(ds: str | None, id_map: dict[str, str]) -> str | None: if not ds or ":" not in ds: return ds kind, _, old_id = ds.partition(":") if kind in ("master", "form"): return f"{kind}:{_remap(old_id, id_map)}" return ds # "dataset:" (Datasets aren't a bundle type) left as-is def _remap_page_links(document: str, id_map: dict[str, str]) -> str: """An html page's "ff-page:" links to other pages in the bundle.""" return re.sub(r"ff-page:([0-9a-fA-F-]{36})", lambda m: f"ff-page:{_remap(m.group(1).lower(), id_map)}", document) def _remap_page_event(event: dict, id_map: dict[str, str]) -> dict: event = dict(event) for key in ("task_id", "workflow_id"): if event.get(key): event[key] = _remap(event[key], id_map) if event.get("target_type") in _BUNDLE_TARGET_TYPES and event.get("target_id"): event["target_id"] = _remap(event["target_id"], id_map) return event def _remap_page_block(block: dict, id_map: dict[str, str]) -> dict: block = dict(block) if block.get("data_source"): block["data_source"] = _remap_data_source(block["data_source"], id_map) for action_key in ("action", "row_click_action"): action = block.get(action_key) if action and action.get("target_type") in _BUNDLE_TARGET_TYPES and action.get("target_id"): action = dict(action) action["target_id"] = _remap(action["target_id"], id_map) block[action_key] = action if block.get("children"): block["children"] = [_remap_page_block(c, id_map) for c in block["children"]] if block.get("groups"): block["groups"] = [ {**g, "children": [_remap_page_block(c, id_map) for c in g.get("children", [])]} for g in block["groups"] ] return block async def _existing_targets(db: AsyncSession, tenant_id: uuid.UUID, kind: str, ids: list[uuid.UUID]) -> set[str]: """Which of these ids still exist in this tenant, for one bundle type.""" if not ids: return set() column = { "master": (Master.id, Master.tenant_id), "form": (Form.form_id, Form.tenant_id), "workflow": (Workflow.workflow_id, Workflow.tenant_id), "task": (Task.id, Task.tenant_id), "page": (Page.id, Page.tenant_id), "dashboard": (Dashboard.id, Dashboard.tenant_id), "menu": (Menu.id, Menu.tenant_id), }[kind] rows = await db.execute(select(column[0]).where(column[1] == tenant_id, column[0].in_(ids)).distinct()) return {str(i) for i in rows.scalars()} async def _draft_version(db: AsyncSession, model, id_column, tenant_id: uuid.UUID, item_id: uuid.UUID, created_by: uuid.UUID | None): """The draft to write an updated Form/Workflow into: the current version if it's still a draft, else a new draft version after it — what's live keeps serving until someone publishes the update in this environment.""" current = ( await db.execute(select(model).where(model.tenant_id == tenant_id, id_column == item_id, model.is_current.is_(True))) ).scalar_one() if current.status == "draft": return current current.is_current = False await db.flush() draft = model(tenant_id=tenant_id, version=current.version + 1, status="draft", is_current=True, created_by=created_by) setattr(draft, id_column.key, item_id) db.add(draft) return draft async def import_bundle(db: AsyncSession, tenant_id: uuid.UUID, bundle: dict, created_by: uuid.UUID | None) -> AppImportResult: """Creates the bundle's items — or, for items an earlier import of the same source environment created here (app_import_links), updates them: promoting dev → test → prod re-imports the same app, and each environment keeps one copy of each item rather than gaining a duplicate per release. Updated Forms/Workflows get a new draft version and Pages a new working copy, so nothing changes for users until it's published here; Masters, Tasks, Dashboards and Menus have no drafts and change in place.""" items: dict[str, list[dict]] = bundle.get("items", {}) created: dict[str, int] = {t: 0 for t in _TYPES} updated: dict[str, int] = {t: 0 for t in _TYPES} warnings: list[str] = [] source_tenant = bundle.get("source_tenant_id") try: source_tenant_id = uuid.UUID(str(source_tenant)) if source_tenant else None except ValueError: source_tenant_id = None links: dict[tuple[str, str], AppImportLink] = {} if source_tenant_id is not None and source_tenant_id != tenant_id: rows = await db.execute( select(AppImportLink).where(AppImportLink.tenant_id == tenant_id, AppImportLink.source_tenant_id == source_tenant_id) ) links = {(link.item_type, str(link.source_id)): link for link in rows.scalars()} # Old id -> id in this tenant: the item an earlier import created (when it # still exists), else a fresh one. id_map: dict[str, str] = {} existing: set[tuple[str, str]] = set() for kind, type_items in items.items(): linked = {item["id"]: str(links[(kind, item["id"])].target_id) for item in type_items if (kind, item["id"]) in links} alive = await _existing_targets(db, tenant_id, kind, [uuid.UUID(t) for t in linked.values()]) if linked else set() for item in type_items: target = linked.get(item["id"]) if target in alive: id_map[item["id"]] = target existing.add((kind, item["id"])) else: id_map[item["id"]] = str(uuid.uuid4()) def is_update(kind: str, item: dict) -> bool: return (kind, item["id"]) in existing def count(kind: str, item: dict) -> None: (updated if is_update(kind, item) else created)[kind] += 1 # Masters first — Forms/Tasks/other Masters may reference them. for m in items.get("master", []): field_schema = [] for f in m.get("field_schema", []): f = dict(f) if f.get("type") == "reference" and f.get("reference_master_id"): f["reference_master_id"] = _remap(f["reference_master_id"], id_map) if f.get("type") == "reference" and f.get("reference_form_id"): f["reference_form_id"] = _remap(f["reference_form_id"], id_map) field_schema.append(f) if is_update("master", m): master = await db.get(Master, uuid.UUID(id_map[m["id"]])) master.name, master.field_schema = m["name"], field_schema else: db.add(Master(id=uuid.UUID(id_map[m["id"]]), tenant_id=tenant_id, name=m["name"], field_schema=field_schema, created_by=created_by)) count("master", m) form_versions: dict[str, int] = {} for f in items.get("form", []): field_definitions = [] for fld in f.get("field_definitions", []): fld = dict(fld) if fld.get("master_id"): fld["master_id"] = _remap(fld["master_id"], id_map) if fld.get("access_rules"): # Roles don't carry across tenants. fld["access_rules"] = [{**r, "role_ids": []} for r in fld["access_rules"] if not r.get("role_ids")] field_definitions.append(fld) events = [] for ev in f.get("events", []): ev = dict(ev) if ev.get("task_id"): ev["task_id"] = _remap(ev["task_id"], id_map) if ev.get("workflow_id"): ev["workflow_id"] = _remap(ev["workflow_id"], id_map) if ev.get("target_type") in _BUNDLE_TARGET_TYPES and ev.get("target_id"): ev["target_id"] = _remap(ev["target_id"], id_map) events.append(ev) new_id = uuid.UUID(id_map[f["id"]]) if is_update("form", f): form = await _draft_version(db, Form, Form.form_id, tenant_id, new_id, created_by) else: form = Form(form_id=new_id, version=1, tenant_id=tenant_id, status="draft", is_current=True, created_by=created_by) db.add(form) form.name, form.field_definitions, form.events, form.prefill = f["name"], field_definitions, events, f.get("prefill") form_versions[str(new_id)] = form.version count("form", f) for t in items.get("task", []): nodes = [] node_had_connector = False node_had_recipients = False for n in t.get("nodes", []): n = dict(n) if n.get("master_id"): n["master_id"] = _remap(n["master_id"], id_map) if n.get("form_id"): n["form_id"] = _remap(n["form_id"], id_map) if "connector_id" in n and n["connector_id"]: n["connector_id"] = None node_had_connector = True if n.get("notify_role_ids") or n.get("notify_user_ids"): n["notify_role_ids"], n["notify_user_ids"] = [], [] node_had_recipients = True nodes.append(n) if is_update("task", t): task = await db.get(Task, uuid.UUID(id_map[t["id"]])) task.name, task.nodes = t["name"], nodes warnings.append(f"Task \"{t['name']}\" was updated in place — tasks have no drafts, so the change is already active.") else: db.add( Task( id=uuid.UUID(id_map[t["id"]]), tenant_id=tenant_id, name=t["name"], nodes=nodes, variables=t.get("variables") or [], status="draft", created_by=created_by, ) ) count("task", t) if node_had_connector: warnings.append(f"Task \"{t['name']}\": one or more node connectors were cleared — re-select them in the Task Designer.") if node_had_recipients: warnings.append(f"Task \"{t['name']}\": notification recipient roles/users were cleared (they don't carry across tenants) — re-select them.") for wf in items.get("workflow", []): steps = [] wf_had_connector = False wf_had_approvers = False wf_had_recipients = False for s in wf.get("step_definitions", []): s = dict(s) if s.get("target_type") in ("master", "form") and s.get("target_id"): s["target_id"] = _remap(s["target_id"], id_map) if s.get("approver_users"): s["approver_users"] = [] wf_had_approvers = True # A Sub-workflow's workflow and a service step's Task, when they're in the bundle too. if (s.get("subprocess") or {}).get("workflow_id"): s["subprocess"] = {**s["subprocess"], "workflow_id": _remap(str(s["subprocess"]["workflow_id"]), id_map)} if (s.get("service") or {}).get("task_id"): s["service"] = {**s["service"], "task_id": _remap(str(s["service"]["task_id"]), id_map)} esc = s.get("escalation") or {} if esc.get("user_ids") or esc.get("role_ids"): s["escalation"] = {**esc, "user_ids": [], "role_ids": []} wf_had_approvers = True notifications = [] for note in s.get("notifications", []): note = dict(note) if note.get("connector_id"): note["connector_id"] = None wf_had_connector = True if note.get("recipient_role_ids") or note.get("recipient_user_ids"): note["recipient_role_ids"], note["recipient_user_ids"] = [], [] wf_had_recipients = True notifications.append(note) s["notifications"] = notifications steps.append(s) new_workflow_id = uuid.UUID(id_map[wf["id"]]) if is_update("workflow", wf): workflow = await _draft_version(db, Workflow, Workflow.workflow_id, tenant_id, new_workflow_id, created_by) await db.flush() await db.execute( sa_delete(WorkflowStepForm).where(WorkflowStepForm.workflow_id == new_workflow_id, WorkflowStepForm.workflow_version == workflow.version) ) else: workflow = Workflow(workflow_id=new_workflow_id, version=1, tenant_id=tenant_id, status="draft", is_current=True, created_by=created_by) db.add(workflow) workflow.name, workflow.step_definitions = wf["name"], steps workflow.settings = dict(wf.get("settings") or {}) count("workflow", wf) if wf_had_connector: warnings.append(f"Workflow \"{wf['name']}\": one or more step-notification connectors were cleared — re-select them in the Workflow Designer.") if wf_had_approvers: warnings.append(f"Workflow \"{wf['name']}\": approver users and escalation people were cleared (users don't carry across tenants) — re-assign them.") if wf_had_recipients: warnings.append(f"Workflow \"{wf['name']}\": notification recipient roles/users were cleared (they don't carry across tenants) — re-select them.") await db.flush() for sf in wf.get("step_forms", []): new_form_id = id_map.get(sf["form_id"]) if new_form_id is None: warnings.append( f"Workflow \"{wf['name']}\" step \"{sf['step_key']}\": its linked form wasn't included in this App, so the step-form link was skipped." ) continue db.add( WorkflowStepForm( workflow_id=new_workflow_id, workflow_version=workflow.version, step_key=sf["step_key"], form_id=uuid.UUID(new_form_id), form_version=form_versions.get(new_form_id, 1), ) ) for p in items.get("page", []): content = { "name": p["name"], "blocks": [_remap_page_block(b, id_map) for b in p.get("blocks", [])], "parameters": p.get("parameters", []), "events": [_remap_page_event(e, id_map) for e in p.get("events", [])], "canvas_width": p.get("canvas_width", 1200), "fit_mode": p.get("fit_mode", "scale"), # An html page's document, re-sanitized (it came from elsewhere), # its links to other imported pages pointed at their ids here. # Whether it's on the public website, and at what address, is # this environment's own call — not carried over. "content_mode": p.get("content_mode") if p.get("content_mode") in ("blocks", "html") else "blocks", "html_document": _remap_page_links(sanitize_document(p["html_document"]), id_map) if p.get("html_document") else None, "seo": p.get("seo") or {}, } if is_update("page", p): # The working copy only — the published version keeps showing # until someone publishes the update here. page = await db.get(Page, uuid.UUID(id_map[p["id"]])) for key, value in content.items(): setattr(page, key, value) else: db.add(Page(id=uuid.UUID(id_map[p["id"]]), tenant_id=tenant_id, status="draft", created_by=created_by, **content)) count("page", p) for d in items.get("dashboard", []): new_dashboard_id = uuid.UUID(id_map[d["id"]]) if is_update("dashboard", d): dashboard = await db.get(Dashboard, new_dashboard_id) dashboard.name, dashboard.parameters = d["name"], d.get("parameters", []) await db.execute(sa_delete(Widget).where(Widget.dashboard_id == new_dashboard_id)) warnings.append(f"Dashboard \"{d['name']}\" was updated in place — dashboards have no drafts.") else: db.add(Dashboard(id=new_dashboard_id, tenant_id=tenant_id, name=d["name"], parameters=d.get("parameters", []), status="draft", created_by=created_by)) count("dashboard", d) dash_had_connector = False for w in d.get("widgets", []): db.add( Widget( tenant_id=tenant_id, dashboard_id=new_dashboard_id, type=w["type"], data_source=_remap_data_source(w["data_source"], id_map), size=w["size"], color=w["color"], refresh_interval=w["refresh_interval"], connector_id=None, sort_order=w["sort_order"], **({"config": w["config"]} if w.get("config") is not None else {}), ) ) if w.get("connector_id"): dash_had_connector = True if dash_had_connector: warnings.append(f"Dashboard \"{d['name']}\": one or more widget connectors were cleared — re-select them in the Dashboard.") menu_items = items.get("menu", []) role_ids = (await db.execute(select(Role.id).where(Role.tenant_id == tenant_id))).scalars().all() if menu_items else [] for m in menu_items: target_id = m.get("target_id") if m.get("target_type") in _BUNDLE_TARGET_TYPES and target_id: target_id = _remap(target_id, id_map) fields = { "label": m["label"], "icon": m["icon"], "link_type": m["link_type"], "linked_screen": m.get("linked_screen"), "target_type": m.get("target_type"), "target_id": uuid.UUID(target_id) if target_id else None, "group_label": m.get("group_label"), "feature_name": m.get("feature_name"), "feature_status": m.get("feature_status"), "external_link_url": m.get("external_link_url"), "sort_order": m.get("sort_order", 0), "status": m.get("status", "active"), "audience": m.get("audience", "internal"), } new_menu_id = uuid.UUID(id_map[m["id"]]) if is_update("menu", m): menu = await db.get(Menu, new_menu_id) for key, value in fields.items(): setattr(menu, key, value) else: db.add(Menu(id=new_menu_id, tenant_id=tenant_id, **fields)) await db.flush() db.add_all(MenuRole(menu_id=new_menu_id, role_id=rid, visible=False) for rid in role_ids) count("menu", m) for missing in bundle.get("missing_items", []): warnings.append(f"Source item no longer existed at export time and was skipped: {missing}.") # Remember what each source item became here, for the next import. if source_tenant_id is not None and source_tenant_id != tenant_id: for kind, type_items in items.items(): for item in type_items: link = links.get((kind, item["id"])) if link is None: db.add( AppImportLink( tenant_id=tenant_id, source_tenant_id=source_tenant_id, item_type=kind, source_id=uuid.UUID(item["id"]), target_id=uuid.UUID(id_map[item["id"]]), ) ) else: link.target_id = uuid.UUID(id_map[item["id"]]) await db.flush() return AppImportResult(created=created, updated=updated, warnings=warnings)