"""FlowDeck — Collections : crud. Découpe A28 de l'ancien app/routers/collections.py (2 622 lignes, 53 routes) — un module par concern, contrat inchangé. """ from __future__ import annotations import json import logging from fastapi import APIRouter, Body, HTTPException, Request from app.db import get_conn from app.routers.dashboard._common import _get_active_workspace from app.services.automations import fire_event, run_event_sync from app.services.collection_lifecycle import delete_collections from app.services.db_templates import materialize_properties from app.services.permission_manager import PermissionManager from ._common import _apply_template, _require_view, _session_user logger = logging.getLogger(__name__) router = APIRouter(tags=["collections"], prefix="/db") # ── API: List & Create (no path params) ── # # `GET /db` (la page HTML) vit dans `index_page.py` : elle était auparavant # un dump JSON de debug, sans layout — alors qu'elle sert de destination au # bouton « Ouvrir mes bases » de My Tasks. @router.get("/api") def list_collections_api(request: Request): """API: list all collections.""" with get_conn() as conn: rows = conn.execute( "SELECT * FROM collections ORDER BY name" ).fetchall() return {"collections": [dict(r) for r in rows]} @router.post("/api") def create_collection_api(request: Request, body: dict = Body(default={})): """API: create a new collection, optionally from a database template.""" name = body.get("name", "").strip() if not name: raise HTTPException(status_code=400, detail="name is required") description = body.get("description", "") icon = body.get("icon", "📋") gitea_owner = body.get("gitea_owner") gitea_repo = body.get("gitea_repo") schema = body.get("schema", []) is_locked = body.get("is_locked", False) # Workspace d'appartenance : celui du body si fourni et autorisé, sinon le # workspace actif de l'utilisateur. Sans cela la collection reste orpheline # et n'apparaît dans aucune vue scopée (ex. /my-tasks). user = _session_user(request) workspace_id = body.get("workspace_id") if workspace_id is not None: try: workspace_id = int(workspace_id) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="workspace_id must be an integer") from None with get_conn() as conn: exists = conn.execute("SELECT 1 FROM workspaces WHERE id=?", (workspace_id,)).fetchone() if not exists: raise HTTPException(status_code=400, detail="workspace_id does not exist") else: active = _get_active_workspace(request, (user or {}).get("id")) workspace_id = active["id"] if active else None with get_conn() as conn: # Apply a template if requested (provides schema + icon). tpl = _apply_template(conn, body.get("template")) if tpl: if body.get("name"): name = body["name"].strip() description = tpl["description"] icon = tpl.get("icon") or icon try: schema = json.loads(tpl["schema_json"]) except (json.JSONDecodeError, TypeError): schema = [] schema_json = json.dumps(schema) cur = conn.execute( """INSERT INTO collections (name, description, icon, schema_json, gitea_owner, gitea_repo, is_locked, workspace_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", (name, description, icon, schema_json, gitea_owner, gitea_repo, int(is_locked), workspace_id), ) collection_id = cur.lastrowid materialize_properties(conn, collection_id, schema) conn.execute( """INSERT INTO collection_views (collection_id, name, view_type, config_json) VALUES (?, ?, ?, ?)""", (collection_id, "Default View", "table", json.dumps({ "visible_properties": ["Title"], "sorts": [], "filters": [], })), ) conn.commit() run_event_sync(fire_event("collection.created", {"collection_id": collection_id, "name": name, "icon": icon})) from app.services.agent_evenements import emettre, url_collection emettre("flowdeck.collection.created", f"collection:{collection_id}", name, url=url_collection(collection_id), actor="user") return {"id": collection_id, "name": name, "status": "created"} # ── API: Update & Delete (no path-param conflicts) ── # ── API: Update & Delete (no path-param conflicts) ── @router.put("/api/{collection_id}") def update_collection_api(request: Request, collection_id: int, body: dict = Body(default={})): """API: update a collection.""" with get_conn() as conn: existing = conn.execute( "SELECT * FROM collections WHERE id=?", (collection_id,) ).fetchone() if not existing: raise HTTPException(status_code=404, detail="Collection not found") name = body.get("name", existing["name"]) description = body.get("description", existing["description"]) icon = body.get("icon", existing["icon"]) is_locked = body.get("is_locked", existing["is_locked"]) schema_json = json.dumps(body.get("schema", json.loads(existing["schema_json"]))) gitea_owner = body.get("gitea_owner", existing["gitea_owner"]) gitea_repo = body.get("gitea_repo", existing["gitea_repo"]) conn.execute( """UPDATE collections SET name=?, description=?, icon=?, schema_json=?, is_locked=?, gitea_owner=?, gitea_repo=?, updated_at=CURRENT_TIMESTAMP WHERE id=?""", (name, description, icon, schema_json, int(is_locked), gitea_owner, gitea_repo, collection_id), ) conn.commit() run_event_sync(fire_event("collection.updated", {"collection_id": collection_id, "name": name})) return {"id": collection_id, "status": "updated"} @router.delete("/api/{collection_id}") def delete_collection_api(request: Request, collection_id: int): """API: delete a collection and its pages (CASCADE).""" # v6.0.0: granular collection permissions — owner/admin only. user = _session_user(request) _require_view(collection_id, user) if user: pm = PermissionManager(user["id"]) if not pm.can_manage_collection_permissions(collection_id): raise HTTPException(status_code=403, detail="Only a collection owner can delete it") with get_conn() as conn: existing = conn.execute( "SELECT * FROM collections WHERE id=?", (collection_id,) ).fetchone() if not existing: raise HTTPException(status_code=404, detail="Collection not found") # `delete_collections` détache la page hôte et les vues liées avant la # suppression : `pages.collection_id` et # `collection_data_sources.source_collection_id` sont en NO ACTION, la # suppression brute levait un IntegrityError. delete_collections(conn, [collection_id]) conn.commit() run_event_sync(fire_event("collection.deleted", {"collection_id": collection_id, "name": existing["name"] if existing else ""})) return {"id": collection_id, "status": "deleted"} @router.post("/{collection_id}/duplicate") def duplicate_collection_api(request: Request, collection_id: int): """v5.4.0: deep-duplicate a database (views + properties + pages + data sources) into a new collection named ' (copy)'.""" with get_conn() as conn: src = conn.execute( "SELECT * FROM collections WHERE id=?", (collection_id,) ).fetchone() if not src: raise HTTPException(status_code=404, detail="Collection not found") new_name = (src["name"] or "Database") + " copy" cur = conn.execute( """INSERT INTO collections (name, description, icon, schema_json, gitea_owner, gitea_repo, is_locked, is_inline, parent_page_id, workspace_id, created_by, is_task, updated_at) SELECT ?, description, icon, schema_json, gitea_owner, gitea_repo, is_locked, is_inline, parent_page_id, workspace_id, created_by, is_task, updated_at FROM collections WHERE id=?""", (new_name, collection_id), ) new_id = cur.lastrowid # ── Properties (remap ids so relation/rollup refs stay valid) ── prop_map: dict[int, int] = {} rows = conn.execute( "SELECT * FROM collection_properties WHERE collection_id=? ORDER BY position", (collection_id,), ).fetchall() # A23 : un seul executemany ; les rowid sont contigus (même transaction, # insertion dans l'ordre de `rows`), donc le mappeur se fait par index. tuples = [ (new_id, p["name"], p["prop_type"], p["options_json"], p["number_format"], None, p["reverse_name"], None, None, p["rollup_function"], p["formula_expression"], p["position"], p["required"], p["visible_in_views"]) for p in rows ] if tuples: ncur = conn.executemany( """INSERT INTO collection_properties (collection_id, name, prop_type, options_json, number_format, related_collection_id, reverse_name, relation_property_id, target_property_id, rollup_function, formula_expression, position, required, visible_in_views) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", tuples, ) new_ids = [ r["id"] for r in conn.execute( "SELECT id FROM collection_properties WHERE collection_id=? ORDER BY id", (new_id,), ).fetchall() ] assert len(new_ids) == len(tuples), "remap des propriétés : effectif inattendu" for p, new_pid in zip(rows, new_ids, strict=True): prop_map[p["id"]] = new_pid # Fix cross-property references after all rows exist (creates may target # columns not inserted yet). Related collection remapped to the copy. for p in rows: p = dict(p) # convert sqlite3.Row to dict new_pid = prop_map[p["id"]] related = p["related_collection_id"] related_new = new_id if related == collection_id else related if p["prop_type"] == "relation": conn.execute( "UPDATE collection_properties SET related_collection_id=? WHERE id=?", (related_new, new_pid), ) if p.get("relation_property_id") and p["relation_property_id"] in prop_map: conn.execute( "UPDATE collection_properties SET relation_property_id=? WHERE id=?", (prop_map[p["relation_property_id"]], new_pid), ) if p.get("target_property_id") and p["target_property_id"] in prop_map: conn.execute( "UPDATE collection_properties SET target_property_id=? WHERE id=?", (prop_map[p["target_property_id"]], new_pid), ) # ── Views ── vrows = conn.execute( "SELECT * FROM collection_views WHERE collection_id=? ORDER BY position", (collection_id,), ).fetchall() for v in vrows: conn.execute( """INSERT INTO collection_views (collection_id, name, view_type, config_json, position) VALUES (?,?,?,?,?)""", (new_id, v["name"], v["view_type"], v["config_json"], v["position"]), ) # ── Pages (rows) with property ids remapped to the copy's properties ── prows = conn.execute( "SELECT * FROM collection_pages WHERE collection_id=? ORDER BY position", (collection_id,), ).fetchall() page_map: dict[int, int] = {} for p in prows: try: pv = json.loads(p["property_values_json"]) if p["property_values_json"] else {} except (json.JSONDecodeError, TypeError): pv = {} pv_new = {} for k, val in pv.items(): try: prop_id = int(k) except (ValueError, TypeError): prop_id = None new_key = str(prop_map.get(prop_id, prop_id)) if prop_id is not None else k pv_new[new_key] = val ncur = conn.execute( """INSERT INTO collection_pages (collection_id, title, icon, position, parent_id, gitea_issue_id, gitea_issue_number, property_values_json, created_at, updated_at) SELECT ?, title, icon, position, NULL, NULL, NULL, ?, created_at, updated_at FROM collection_pages WHERE id=?""", (new_id, json.dumps(pv_new), p["id"]), ) page_map[p["id"]] = ncur.lastrowid # Re-parent sub-items to the copied rows. for p in prows: if p["parent_id"] and p["parent_id"] in page_map: conn.execute( "UPDATE collection_pages SET parent_id=? WHERE id=?", (page_map[p["parent_id"]], page_map[p["id"]]), ) # ── Data sources (linked DBs) ── drows = conn.execute( "SELECT * FROM collection_data_sources WHERE collection_id=?", (collection_id,), ).fetchall() for d in drows: src_coll = d["source_collection_id"] src_now = new_id if src_coll == collection_id else src_coll conn.execute( """INSERT INTO collection_data_sources (collection_id, source_collection_id, source_name, is_linked, position) VALUES (?,?,?,?,?)""", (new_id, src_now, d["source_name"], d["is_linked"], d["position"]), ) conn.commit() run_event_sync(fire_event("collection.created", {"collection_id": new_id, "name": new_name})) return {"id": new_id, "name": new_name, "status": "duplicated"} # ── Page CRUD (standalone, BEFORE collection wildcards) ──