feat: create a new canvas by copying an existing one

Add a 'Copy from existing' option to the New Canvas modal. It lists every
canvas with node/group/text counts; picking one deep-copies its nodes, edges,
parent/child links and canvas state (viewport, custom style, floor plan) into
a fresh design.

Backend: POST /designs/{source_id}/copy remaps node ids, re-points edges and
parent links, and clones canvas state; GET /designs now returns per-design
counts for the picker. Standalone mode clones the localStorage canvas.

Closes #216

ha-relevant: maybe
This commit is contained in:
Pouzor
2026-07-07 15:53:42 +02:00
parent 6b591cdd88
commit 0b4bd5680d
10 changed files with 493 additions and 8 deletions
+101 -3
View File
@@ -1,14 +1,21 @@
import uuid
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_current_user
from app.db.database import get_db
from app.db.models import CanvasState, Design, Edge, Node
from app.schemas.designs import DesignCreate, DesignResponse, DesignUpdate
from app.schemas.designs import DesignCopy, DesignCreate, DesignResponse, DesignUpdate
router = APIRouter()
# Node.type values that are canvas annotations rather than real devices. Kept in
# sync with the frontend (Sidebar counts, canvasSerializer types).
_GROUP_TYPE = "groupRect"
_TEXT_TYPE = "text"
@router.get("", response_model=list[DesignResponse])
async def list_designs(
@@ -16,7 +23,31 @@ async def list_designs(
_: str = Depends(get_current_user),
) -> list[DesignResponse]:
designs = (await db.execute(select(Design).order_by(Design.created_at))).scalars().all()
return [DesignResponse.model_validate(d) for d in designs]
# One grouped query for all designs → node/group/text counts per design.
rows = (
await db.execute(select(Node.design_id, Node.type, func.count()).group_by(Node.design_id, Node.type))
).all()
counts: dict[str, dict[str, int]] = {}
for design_id, node_type, count in rows:
if design_id is None:
continue
bucket = counts.setdefault(design_id, {"node": 0, "group": 0, "text": 0})
if node_type == _GROUP_TYPE:
bucket["group"] += count
elif node_type == _TEXT_TYPE:
bucket["text"] += count
else:
bucket["node"] += count
result = []
for d in designs:
resp = DesignResponse.model_validate(d)
c = counts.get(d.id, {"node": 0, "group": 0, "text": 0})
resp.node_count = c["node"]
resp.group_count = c["group"]
resp.text_count = c["text"]
result.append(resp)
return result
@router.post("", response_model=DesignResponse, status_code=201)
@@ -35,6 +66,73 @@ async def create_design(
return DesignResponse.model_validate(design)
@router.post("/{source_id}/copy", response_model=DesignResponse, status_code=201)
async def copy_design(
source_id: str,
body: DesignCopy,
db: AsyncSession = Depends(get_db),
_: str = Depends(get_current_user),
) -> DesignResponse:
"""Create a new design that deep-copies the source's nodes, edges and canvas state."""
source = await db.get(Design, source_id)
if not source:
raise HTTPException(404, "Source design not found")
new_design = Design(name=body.name, icon=body.icon, design_type=source.design_type)
db.add(new_design)
await db.flush()
src_nodes = (await db.execute(select(Node).where(Node.design_id == source_id))).scalars().all()
src_edges = (await db.execute(select(Edge).where(Edge.design_id == source_id))).scalars().all()
# New id per source node so edges and parent links can be re-pointed at the copy.
id_map = {n.id: str(uuid.uuid4()) for n in src_nodes}
# Columns we set explicitly or let the DB default — never copy verbatim.
node_skip = {"id", "design_id", "parent_id", "created_at", "updated_at"}
for n in src_nodes:
cols = {c.name: getattr(n, c.name) for c in Node.__table__.columns if c.name not in node_skip}
db.add(Node(id=id_map[n.id], design_id=new_design.id, parent_id=None, **cols))
await db.flush() # nodes must exist before we wire self-referential parent_id
# Second pass: re-point parent links inside the copy.
for n in src_nodes:
if n.parent_id and n.parent_id in id_map:
child = await db.get(Node, id_map[n.id])
if child:
child.parent_id = id_map[n.parent_id]
edge_skip = {"id", "design_id", "source", "target", "created_at"}
for e in src_edges:
# Skip edges whose endpoints aren't part of this design (dangling FKs).
if e.source not in id_map or e.target not in id_map:
continue
cols = {c.name: getattr(e, c.name) for c in Edge.__table__.columns if c.name not in edge_skip}
db.add(
Edge(
id=str(uuid.uuid4()),
design_id=new_design.id,
source=id_map[e.source],
target=id_map[e.target],
**cols,
)
)
# Copy canvas state (viewport, custom style, and the floor plan carried in viewport).
src_state = await db.get(CanvasState, source_id)
db.add(
CanvasState(
design_id=new_design.id,
viewport=src_state.viewport if src_state else {},
custom_style=src_state.custom_style if src_state else None,
)
)
await db.commit()
await db.refresh(new_design)
return DesignResponse.model_validate(new_design)
@router.put("/{design_id}", response_model=DesignResponse)
async def update_design(
design_id: str,
+12
View File
@@ -16,6 +16,13 @@ class DesignUpdate(BaseModel):
icon: str | None = None
class DesignCopy(BaseModel):
"""Create a new design by deep-copying an existing one's canvas."""
name: str
icon: str = "dashboard"
class DesignResponse(BaseModel):
id: str
name: str
@@ -23,5 +30,10 @@ class DesignResponse(BaseModel):
icon: str | None = None
created_at: datetime
updated_at: datetime
# Populated by list_designs so the "copy from existing" picker can show what
# each canvas holds. None on create/update/copy responses (not computed there).
node_count: int | None = None
group_count: int | None = None
text_count: int | None = None
model_config = {"from_attributes": True}
+112
View File
@@ -100,6 +100,118 @@ async def test_list_returns_created_designs_ordered(client: AsyncClient, headers
assert ids == [a["id"], b["id"]]
# ── counts in list ──────────────────────────────────────────────────────────
async def test_list_includes_node_group_text_counts(client: AsyncClient, headers: dict):
design = await _create(client, headers, name="Counted")
server = node_payload(label="S", type="server")
group = node_payload(label="G", type="groupRect")
text = node_payload(label="T", type="text")
save = await client.post(
"/api/v1/canvas/save",
json={"nodes": [server, group, text], "edges": [], "viewport": {}, "design_id": design["id"]},
headers=headers,
)
assert save.status_code == 200
listed = (await client.get("/api/v1/designs", headers=headers)).json()
d = next(x for x in listed if x["id"] == design["id"])
assert d["node_count"] == 1
assert d["group_count"] == 1
assert d["text_count"] == 1
async def test_list_counts_zero_for_empty_design(client: AsyncClient, headers: dict):
design = await _create(client, headers, name="Empty")
listed = (await client.get("/api/v1/designs", headers=headers)).json()
d = next(x for x in listed if x["id"] == design["id"])
assert d["node_count"] == 0
assert d["group_count"] == 0
assert d["text_count"] == 0
# ── copy ──────────────────────────────────────────────────────────────────────
async def test_copy_requires_auth(client: AsyncClient):
res = await client.post(f"/api/v1/designs/{uuid.uuid4()}/copy", json={"name": "X"})
assert res.status_code == 401
async def test_copy_missing_source_returns_404(client: AsyncClient, headers: dict):
res = await client.post(f"/api/v1/designs/{uuid.uuid4()}/copy", json={"name": "X"}, headers=headers)
assert res.status_code == 404
async def test_copy_duplicates_nodes_edges_and_remaps_ids(client: AsyncClient, headers: dict):
source = await _create(client, headers, name="Source", icon="server")
n1 = node_payload(label="A")
n2 = node_payload(label="B")
e1 = edge_payload(n1["id"], n2["id"], label="link")
save = await client.post(
"/api/v1/canvas/save",
json={"nodes": [n1, n2], "edges": [e1], "viewport": {"x": 5, "y": 6, "zoom": 2}, "design_id": source["id"]},
headers=headers,
)
assert save.status_code == 200
res = await client.post(
f"/api/v1/designs/{source['id']}/copy", json={"name": "Copy", "icon": "network"}, headers=headers,
)
assert res.status_code == 201, res.text
copy = res.json()
assert copy["name"] == "Copy"
assert copy["icon"] == "network"
assert copy["design_type"] == "network"
assert copy["id"] != source["id"]
# Copied canvas has the same shape but fresh node ids, and edge re-pointed.
canvas = (await client.get("/api/v1/canvas", params={"design_id": copy["id"]}, headers=headers)).json()
assert {n["label"] for n in canvas["nodes"]} == {"A", "B"}
copied_ids = {n["id"] for n in canvas["nodes"]}
assert copied_ids.isdisjoint({n1["id"], n2["id"]})
assert len(canvas["edges"]) == 1
edge = canvas["edges"][0]
assert edge["source"] in copied_ids
assert edge["target"] in copied_ids
assert edge["label"] == "link"
assert canvas["viewport"] == {"x": 5, "y": 6, "zoom": 2}
async def test_copy_remaps_parent_child_relationship(client: AsyncClient, headers: dict):
source = await _create(client, headers, name="Nested")
parent = node_payload(label="P", type="proxmox", container_mode=True)
child = node_payload(label="C", type="vm", parent_id=parent["id"])
save = await client.post(
"/api/v1/canvas/save",
json={"nodes": [parent, child], "edges": [], "viewport": {}, "design_id": source["id"]},
headers=headers,
)
assert save.status_code == 200
res = await client.post(f"/api/v1/designs/{source['id']}/copy", json={"name": "Copy"}, headers=headers)
assert res.status_code == 201
copy = res.json()
canvas = (await client.get("/api/v1/canvas", params={"design_id": copy["id"]}, headers=headers)).json()
by_label = {n["label"]: n for n in canvas["nodes"]}
# Child's parent_id points at the COPIED parent, not the original.
assert by_label["C"]["parent_id"] == by_label["P"]["id"]
assert by_label["C"]["parent_id"] != parent["id"]
async def test_copy_leaves_source_untouched(client: AsyncClient, headers: dict):
source = await _create(client, headers, name="Source")
n1 = node_payload(label="A")
await client.post(
"/api/v1/canvas/save",
json={"nodes": [n1], "edges": [], "viewport": {}, "design_id": source["id"]},
headers=headers,
)
await client.post(f"/api/v1/designs/{source['id']}/copy", json={"name": "Copy"}, headers=headers)
src_canvas = (await client.get("/api/v1/canvas", params={"design_id": source["id"]}, headers=headers)).json()
assert len(src_canvas["nodes"]) == 1
assert src_canvas["nodes"][0]["id"] == n1["id"]
# ── update ────────────────────────────────────────────────────────────────────
async def test_update_design_renames(client: AsyncClient, headers: dict):