"""FlowDeck — Public API v2 : collections. Découpe A28 de l'ancien app/routers/api_v2.py (2 110 lignes, 115 routes) — un module par concern, contrat inchangé (Bearer+scopes, pagination, RFC7807, audit + idempotency). """ from __future__ import annotations import json import logging from fastapi import APIRouter, Body, Header, HTTPException, Request from fastapi.responses import JSONResponse from app.db import get_conn from app.services.api_v2_helpers import ( # noqa: F401 — require_scope est utilisé par les handlers audit_log, check_idempotency, check_v2_rate_limit, get_bearer_user, has_scope, paginate_headers, parse_pagination, require_scope, row_to_dict, store_idempotency, to_iso8601, validate_scopes_input, ) from app.services.automations import fire_event as _fire_event from app.services.automations import run_event_sync from app.services.collection_lifecycle import delete_collections from ._common import _v2_rate_check logger = logging.getLogger(__name__) router = APIRouter(tags=["api-v2"]) @router.get("/collections") def list_collections_v2(request: Request, authorization: str | None = Header(default=None)): user = get_bearer_user(request, authorization) _v2_rate_check(request, user) limit, offset = parse_pagination(request) ws_filter = request.query_params.get("workspace_id") q = (request.query_params.get("query") or "").strip() with get_conn() as conn: where = [] params: list = [] if ws_filter: try: wid = int(ws_filter) where.append("c.workspace_id=?") params.append(wid) except ValueError: pass if q: where.append("(c.name LIKE ? OR c.description LIKE ?)") like = f"%{q}%" params.extend([like, like]) clause = ("WHERE " + " AND ".join(where)) if where else "" total = conn.execute(f"SELECT COUNT(*) FROM collections c {clause}", params).fetchone()[0] rows = conn.execute(f"SELECT c.* FROM collections c {clause} ORDER BY c.name LIMIT ? OFFSET ?", (*params, limit, offset)).fetchall() cols = [] for r in rows: d = row_to_dict(r) # filter by visibility: skip private not visible (best-effort) cols.append(d) resp = {"collections": cols, "total": total, "limit": limit, "offset": offset} return JSONResponse(content=resp, headers=paginate_headers(total)) @router.post("/collections") def create_collection_v2(request: Request, authorization: str | None = Header(default=None), body: dict = Body(default={})): user = require_scope("write")(request, authorization) _v2_rate_check(request, user) idem = check_idempotency(request, user["id"]) if idem: return JSONResponse(content=idem["data"], status_code=idem["status"]) name = (body.get("name") or "").strip() if not name: raise HTTPException(400, "name is required") description = body.get("description", "") icon = body.get("icon", "📋") workspace_id = body.get("workspace_id") schema = body.get("schema") or body.get("schema_json") or [] if isinstance(schema, str): try: schema = json.loads(schema) except Exception: schema = [] schema_json = json.dumps(schema) with get_conn() as conn: cur = conn.execute("INSERT INTO collections (name, description, icon, schema_json, workspace_id, created_by) VALUES (?, ?, ?, ?, ?, ?)", (name, description, icon, schema_json, workspace_id, user["id"])) cid = cur.lastrowid # materialize properties if schema provided — A25 : PAS de try ici, # une exception doit interrompre la transaction (sinon la collection est # commitée sans son schéma et l'erreur disparaît). from app.services.db_templates import materialize_properties materialize_properties(conn, cid, schema) # default view conn.execute("INSERT INTO collection_views (collection_id, name, view_type, config_json) VALUES (?,?,?,?)", (cid, "Default View", "table", json.dumps({"visible_properties": ["Title"]}))) conn.commit() row = conn.execute("SELECT * FROM collections WHERE id=?", (cid,)).fetchone() audit_log(user, "collection.create", "collection", cid, name, request) from app.services.agent_evenements import emettre, url_collection emettre("flowdeck.collection.created", f"collection:{cid}", name, url=url_collection(cid), actor="user") data = {"id": cid, "name": name, "status": "created", "collection": row_to_dict(row)} key = (request.headers.get("Idempotency-Key") or "").strip() if key: store_idempotency(key, user["id"], data, 201) return JSONResponse(content=data, status_code=201) @router.get("/collections/{collection_id}") def get_collection_v2(collection_id: int, request: Request, authorization: str | None = Header(default=None)): user = get_bearer_user(request, authorization) _v2_rate_check(request, user) with get_conn() as conn: row = conn.execute("SELECT * FROM collections WHERE id=?", (collection_id,)).fetchone() if not row: raise HTTPException(404, "Collection not found") pages = conn.execute("SELECT id, title, icon, position, property_values_json, created_at FROM collection_pages WHERE collection_id=? ORDER BY position LIMIT 50", (collection_id,)).fetchall() d = row_to_dict(row) d["pages"] = [row_to_dict(p) for p in pages] return d @router.patch("/collections/{collection_id}") def patch_collection_v2(collection_id: int, request: Request, authorization: str | None = Header(default=None), body: dict = Body(default={})): user = require_scope("write")(request, authorization) _v2_rate_check(request, user) with get_conn() as conn: row = conn.execute("SELECT * FROM collections WHERE id=?", (collection_id,)).fetchone() if not row: raise HTTPException(404, "Collection not found") name = body.get("name", row["name"]) description = body.get("description", row["description"]) icon = body.get("icon", row["icon"]) schema = body.get("schema") or body.get("schema_json") if schema is not None: sj = json.dumps(schema) if isinstance(schema, (list, dict)) else str(schema) else: sj = row["schema_json"] conn.execute("UPDATE collections SET name=?, description=?, icon=?, schema_json=?, updated_at=CURRENT_TIMESTAMP WHERE id=?", (name, description, icon, sj, collection_id)) conn.commit() audit_log(user, "collection.update", "collection", collection_id, "", request) return {"id": collection_id, "status": "updated"} @router.delete("/collections/{collection_id}") def delete_collection_v2(collection_id: int, request: Request, authorization: str | None = Header(default=None)): user = require_scope("write")(request, authorization) _v2_rate_check(request, user) with get_conn() as conn: row = conn.execute("SELECT * FROM collections WHERE id=?", (collection_id,)).fetchone() if not row: raise HTTPException(404, "Collection not found") delete_collections(conn, [collection_id]) conn.commit() audit_log(user, "collection.delete", "collection", collection_id, "", request) return {"id": collection_id, "status": "deleted"} @router.post("/collections/{collection_id}/linked") def create_linked_db(collection_id: int, request: Request, authorization: str | None = Header(default=None), body: dict = Body(default={})): user = require_scope("write")(request, authorization) _v2_rate_check(request, user) name = (body.get("name") or "").strip() or f"Linked DB {collection_id}" with get_conn() as conn: src = conn.execute("SELECT * FROM collections WHERE id=?", (collection_id,)).fetchone() if not src: raise HTTPException(404, "Collection not found") cur = conn.execute("INSERT INTO collections (name, description, icon, schema_json, workspace_id, created_by) VALUES (?, ?, ?, ?, ?, ?)", (name, src["description"], src["icon"], src["schema_json"], src["workspace_id"] if "workspace_id" in src.keys() else None, user["id"])) nid = cur.lastrowid # copy data source as linked try: conn.execute("INSERT INTO collection_data_sources (collection_id, source_collection_id, is_linked) VALUES (?, ?, 1)", (nid, collection_id)) except Exception: logger.exception("create_linked_db") # copy views + properties (light) rows = conn.execute("SELECT * FROM collection_properties WHERE collection_id=?", (collection_id,)).fetchall() for p in rows: try: conn.execute("INSERT INTO collection_properties (collection_id, name, prop_type, options_json, position) VALUES (?, ?, ?, ?, ?)", (nid, p["name"], p["prop_type"], p["options_json"], p["position"])) except Exception: logger.exception("create_linked_db") vrows = conn.execute("SELECT * FROM collection_views WHERE collection_id=?", (collection_id,)).fetchall() for v in vrows: try: conn.execute("INSERT INTO collection_views (collection_id, name, view_type, config_json, position) VALUES (?, ?, ?, ?, ?)", (nid, v["name"], v["view_type"], v["config_json"], v["position"])) except Exception: logger.exception("create_linked_db") conn.commit() audit_log(user, "collection.linked", "collection", nid, f"src={collection_id}", request) return {"id": nid, "name": name, "status": "created"} @router.post("/collections/{collection_id}/task") def toggle_task(collection_id: int, request: Request, authorization: str | None = Header(default=None)): user = require_scope("write")(request, authorization) _v2_rate_check(request, user) with get_conn() as conn: row = conn.execute("SELECT is_task FROM collections WHERE id=?", (collection_id,)).fetchone() if not row: raise HTTPException(404, "Collection not found") cur_val = row["is_task"] if "is_task" in row.keys() else 0 new_val = 0 if cur_val else 1 conn.execute("UPDATE collections SET is_task=? WHERE id=?", (new_val, collection_id)) conn.commit() audit_log(user, "collection.toggle_task", "collection", collection_id, str(new_val), request) return {"id": collection_id, "is_task": bool(new_val)} @router.get("/collections/{collection_id}/sources") def list_sources(collection_id: int, request: Request, authorization: str | None = Header(default=None)): user = get_bearer_user(request, authorization) _v2_rate_check(request, user) with get_conn() as conn: if not conn.execute("SELECT id FROM collections WHERE id=?", (collection_id,)).fetchone(): raise HTTPException(404, "Collection not found") rows = conn.execute("SELECT * FROM collection_data_sources WHERE collection_id=? ORDER BY position", (collection_id,)).fetchall() return {"sources": [dict(r) for r in rows]} @router.post("/collections/{collection_id}/sources") def add_source(collection_id: int, request: Request, authorization: str | None = Header(default=None), body: dict = Body(default={})): user = require_scope("write")(request, authorization) _v2_rate_check(request, user) src_id = body.get("source_collection_id") or body.get("source_id") if not src_id: raise HTTPException(400, "source_collection_id required") with get_conn() as conn: if not conn.execute("SELECT id FROM collections WHERE id=?", (collection_id,)).fetchone(): raise HTTPException(404, "Collection not found") if not conn.execute("SELECT id FROM collections WHERE id=?", (src_id,)).fetchone(): raise HTTPException(404, "Source collection not found") try: conn.execute("INSERT INTO collection_data_sources (collection_id, source_collection_id) VALUES (?, ?)", (collection_id, src_id)) conn.commit() except Exception as e: raise HTTPException(409, str(e)) from None audit_log(user, "collection.add_source", "collection", collection_id, str(src_id), request) return {"collection_id": collection_id, "source_collection_id": src_id, "status": "added"} @router.delete("/collections/{collection_id}/sources/{source_id}") def remove_source(collection_id: int, source_id: int, request: Request, authorization: str | None = Header(default=None)): user = require_scope("write")(request, authorization) _v2_rate_check(request, user) with get_conn() as conn: conn.execute("DELETE FROM collection_data_sources WHERE collection_id=? AND (id=? OR source_collection_id=?)", (collection_id, source_id, source_id)) conn.commit() audit_log(user, "collection.remove_source", "collection", collection_id, str(source_id), request) return {"status": "removed"} @router.get("/collections/{collection_id}/pages") def list_collection_pages_v2(collection_id: int, request: Request, authorization: str | None = Header(default=None)): user = get_bearer_user(request, authorization) _v2_rate_check(request, user) limit, offset = parse_pagination(request) with get_conn() as conn: if not conn.execute("SELECT id FROM collections WHERE id=?", (collection_id,)).fetchone(): raise HTTPException(404, "Collection not found") total = conn.execute("SELECT COUNT(*) FROM collection_pages WHERE collection_id=?", (collection_id,)).fetchone()[0] # filters: filter[status]=Done etc., sort, fields # Simple: filter by property name via property_values_json LIKE (best-effort), sort by position or title sort = request.query_params.get("sort") or "" order = "position" desc = False if sort: if sort.startswith("-"): desc = True sort = sort[1:] # allow sorting by title/position/created_at if sort in ("title", "position", "created_at", "updated_at"): order = sort direction = "DESC" if desc else "ASC" rows = conn.execute(f"SELECT * FROM collection_pages WHERE collection_id=? ORDER BY {order} {direction} LIMIT ? OFFSET ?", (collection_id, limit, offset)).fetchall() # apply filter[xxx] in-memory (small) filters = {k[7:-1]: v for k, v in request.query_params.items() if k.startswith("filter[") and k.endswith("]")} fields = request.query_params.get("fields") fields_set = set(fields.split(",")) if fields else None out = [] for r in rows: d = row_to_dict(r) # property filter (AND) if filters: try: pv = json.loads(r["property_values_json"] or "{}") if isinstance(r["property_values_json"], str) else r["property_values_json"] except Exception: pv = {} ok = True for fk, fv in filters.items(): # lookup by prop id or name found = False for kk, vv in (pv or {}).items(): if str(kk) == str(fk) or str(kk).lower() == fk.lower(): if str(vv) == str(fv): found = True break # also check title if filter field is title if fk == "title" and d.get("title") == fv: found = True if not found: ok = False break if not ok: continue if fields_set: d = {k: v for k, v in d.items() if k in fields_set or k in ("id", "collection_id")} out.append(d) return JSONResponse(content={"pages": out, "total": total, "limit": limit, "offset": offset}, headers=paginate_headers(total)) @router.post("/collections/{collection_id}/pages") def create_collection_page_v2(collection_id: int, request: Request, authorization: str | None = Header(default=None), body: dict = Body(default={})): user = require_scope("write")(request, authorization) _v2_rate_check(request, user) idem = check_idempotency(request, user["id"]) if idem: return JSONResponse(content=idem["data"], status_code=idem["status"]) title = (body.get("title") or body.get("name") or "Untitled").strip() or "Untitled" icon = body.get("icon", "file") parent_id = body.get("parent_id") prop_vals = body.get("property_values") or body.get("properties") or body.get("property_values_json") or {} if isinstance(prop_vals, str): try: prop_vals = json.loads(prop_vals) except Exception: prop_vals = {} with get_conn() as conn: if not conn.execute("SELECT id FROM collections WHERE id=?", (collection_id,)).fetchone(): raise HTTPException(404, "Collection not found") # validate properties if helper exists try: pass # light validation: we rely on existing validators except Exception: logger.exception("create_collection_page_v2") max_pos = conn.execute("SELECT COALESCE(MAX(position), -1)+1 FROM collection_pages WHERE collection_id=?", (collection_id,)).fetchone()[0] # apply auto props try: props_list = [dict(r) for r in conn.execute("SELECT * FROM collection_properties WHERE collection_id=?", (collection_id,)).fetchall()] from app.services.property_types import apply_auto_properties as _aap _aap(props_list, prop_vals, user, is_create=True) except Exception: logger.exception("create_collection_page_v2") cur = conn.execute("INSERT INTO collection_pages (collection_id, title, icon, position, parent_id, property_values_json) VALUES (?, ?, ?, ?, ?, ?)", (collection_id, title, icon, max_pos, parent_id, json.dumps(prop_vals))) pid = cur.lastrowid conn.commit() row = conn.execute("SELECT * FROM collection_pages WHERE id=?", (pid,)).fetchone() audit_log(user, "page.create", "collection_page", pid, title, request) from app.services.agent_evenements import emettre, url_collection emettre("flowdeck.page.created", f"page:{pid}", title, url=url_collection(collection_id), actor="user") try: run_event_sync(_fire_event("collection.page.created", {"page_id": pid, "collection_id": collection_id, "title": title})) except Exception: logger.exception("create_collection_page_v2") data = {"id": pid, "title": title, "status": "created", "page": row_to_dict(row)} key = (request.headers.get("Idempotency-Key") or "").strip() if key: store_idempotency(key, user["id"], data, 201) return JSONResponse(content=data, status_code=201) @router.get("/pages/{page_id}") def get_page_v2(page_id: int, request: Request, authorization: str | None = Header(default=None)): user = get_bearer_user(request, authorization) _v2_rate_check(request, user) with get_conn() as conn: row = conn.execute("SELECT * FROM collection_pages WHERE id=?", (page_id,)).fetchone() if not row: # also try pages table (block pages) row2 = conn.execute("SELECT * FROM pages WHERE id=?", (page_id,)).fetchone() if not row2: raise HTTPException(404, "Page not found") d = row_to_dict(row2) # v6.5.0: resolve synced blocks server-side (fresh content). if (d.get("content_format") or "blocks") == "blocks" and d.get("content"): from app.services.synced_blocks import resolve_content_json d["content"] = resolve_content_json(d["content"], d["content_format"]) return d d = row_to_dict(row) # property_values_json already parsed by row_to_dict # v6.5.0: expose the row's content page when it exists (no lazy # creation on a read-only endpoint). content_page_id = conn.execute( "SELECT id FROM pages WHERE collection_row_id=?", (page_id,), ).fetchone() d["content_page_id"] = content_page_id["id"] if content_page_id else None return d @router.patch("/pages/{page_id}") def patch_page_v2(page_id: int, request: Request, authorization: str | None = Header(default=None), body: dict = Body(default={})): user = require_scope("write")(request, authorization) _v2_rate_check(request, user) with get_conn() as conn: row = conn.execute("SELECT * FROM collection_pages WHERE id=?", (page_id,)).fetchone() if not row: raise HTTPException(404, "Page not found") title = body.get("title", row["title"]) icon = body.get("icon", row["icon"]) pos = body.get("position", row["position"]) parent_id = body.get("parent_id", row["parent_id"]) pv_raw = row["property_values_json"] or "{}" try: stored = json.loads(pv_raw) if isinstance(pv_raw, str) else dict(pv_raw) except Exception: stored = {} incoming = body.get("property_values") or body.get("properties") if incoming is not None: if isinstance(incoming, str): try: incoming = json.loads(incoming) except Exception: incoming = {} # merge for k, v in (incoming or {}).items(): stored[str(k)] = v # apply auto props try: props_list = [dict(r) for r in conn.execute("SELECT * FROM collection_properties WHERE collection_id=?", (row["collection_id"],)).fetchall()] from app.services.property_types import apply_auto_properties as _aap _aap(props_list, stored, user, is_create=False) except Exception: logger.exception("patch_page_v2") conn.execute("UPDATE collection_pages SET title=?, icon=?, position=?, parent_id=?, property_values_json=?, updated_at=CURRENT_TIMESTAMP WHERE id=?", (title, icon, pos, parent_id, json.dumps(stored), page_id)) conn.commit() audit_log(user, "page.update", "collection_page", page_id, "", request) from app.services.agent_evenements import emettre, url_collection emettre("flowdeck.page.updated", f"page:{page_id}", title, url=url_collection(row["collection_id"]), actor="user") try: run_event_sync(_fire_event("collection.page.updated", {"page_id": page_id, "collection_id": row["collection_id"], "title": title})) except Exception: logger.exception("patch_page_v2") return {"id": page_id, "status": "updated"} @router.delete("/pages/{page_id}") def delete_page_v2(page_id: int, request: Request, authorization: str | None = Header(default=None)): user = require_scope("write")(request, authorization) _v2_rate_check(request, user) with get_conn() as conn: row = conn.execute("SELECT * FROM collection_pages WHERE id=?", (page_id,)).fetchone() if not row: raise HTTPException(404, "Page not found") conn.execute("DELETE FROM collection_pages WHERE id=?", (page_id,)) conn.commit() audit_log(user, "page.delete", "collection_page", page_id, "", request) from app.services.agent_evenements import emettre, url_collection emettre("flowdeck.page.deleted", f"page:{page_id}", row["title"], url=url_collection(row["collection_id"]), actor="user") try: run_event_sync(_fire_event("collection.page.deleted", {"page_id": page_id, "collection_id": row["collection_id"]})) except Exception: logger.exception("delete_page_v2") return {"id": page_id, "status": "deleted"} @router.post("/pages/{page_id}/restore") def restore_page_v2(page_id: int, request: Request, authorization: str | None = Header(default=None)): user = require_scope("write")(request, authorization) _v2_rate_check(request, user) # For soft-deleted pages (deleted_at) - but collection_pages has no deleted_at; handle pages table with get_conn() as conn: row = conn.execute("SELECT deleted_at FROM pages WHERE id=?", (page_id,)).fetchone() if row and row["deleted_at"]: conn.execute("UPDATE pages SET deleted_at=NULL WHERE id=?", (page_id,)) conn.commit() try: run_event_sync(_fire_event("page.restored", {"page_id": page_id})) except Exception: logger.exception("restore_page_v2") return {"id": page_id, "status": "restored"} raise HTTPException(404, "Page not found or not deleted") @router.post("/pages/{page_id}/move") def move_page_v2(page_id: int, request: Request, authorization: str | None = Header(default=None), body: dict = Body(default={})): user = require_scope("write")(request, authorization) _v2_rate_check(request, user) with get_conn() as conn: row = conn.execute("SELECT * FROM collection_pages WHERE id=?", (page_id,)).fetchone() if not row: raise HTTPException(404, "Page not found") parent_id = body.get("parent_id", row["parent_id"]) position = body.get("position", row["position"]) conn.execute("UPDATE collection_pages SET parent_id=?, position=?, updated_at=CURRENT_TIMESTAMP WHERE id=?", (parent_id, position, page_id)) conn.commit() audit_log(user, "page.move", "collection_page", page_id, f"parent={parent_id} pos={position}", request) return {"id": page_id, "status": "moved"} @router.get("/pages/{page_id}/sub-items") def list_sub_items_v2(page_id: int, request: Request, authorization: str | None = Header(default=None)): user = get_bearer_user(request, authorization) _v2_rate_check(request, user) with get_conn() as conn: if not conn.execute("SELECT id FROM collection_pages WHERE id=?", (page_id,)).fetchone(): raise HTTPException(404, "Page not found") rows = conn.execute("SELECT * FROM collection_pages WHERE parent_id=? ORDER BY position", (page_id,)).fetchall() return {"sub_items": [row_to_dict(r) for r in rows]} @router.post("/pages/{page_id}/sub-items") def create_sub_item_v2(page_id: int, request: Request, authorization: str | None = Header(default=None), body: dict = Body(default={})): user = require_scope("write")(request, authorization) _v2_rate_check(request, user) with get_conn() as conn: parent = conn.execute("SELECT collection_id FROM collection_pages WHERE id=?", (page_id,)).fetchone() if not parent: raise HTTPException(404, "Page not found") title = (body.get("title") or "Untitled").strip() max_pos = conn.execute("SELECT COALESCE(MAX(position), -1)+1 FROM collection_pages WHERE parent_id=?", (page_id,)).fetchone()[0] pv = json.dumps(body.get("property_values") or {}) cur = conn.execute("INSERT INTO collection_pages (collection_id, title, parent_id, position, property_values_json) VALUES (?, ?, ?, ?, ?)", (parent["collection_id"], title, page_id, max_pos, pv)) nid = cur.lastrowid conn.commit() audit_log(user, "page.create_subitem", "collection_page", nid, title, request) return {"id": nid, "status": "created"} @router.get("/pages/{page_id}/dependencies") def list_dependencies_v2(page_id: int, request: Request, authorization: str | None = Header(default=None)): user = get_bearer_user(request, authorization) _v2_rate_check(request, user) with get_conn() as conn: rows = conn.execute("SELECT * FROM page_dependencies WHERE page_id=?", (page_id,)).fetchall() return {"dependencies": [dict(r) for r in rows]} @router.post("/pages/{page_id}/dependencies") def add_dependency_v2(page_id: int, request: Request, authorization: str | None = Header(default=None), body: dict = Body(default={})): user = require_scope("write")(request, authorization) _v2_rate_check(request, user) dep_id = body.get("dependency_id") or body.get("depends_on") dtype = body.get("dependency_type") or "blocks" if not dep_id: raise HTTPException(400, "dependency_id required") with get_conn() as conn: if not conn.execute("SELECT id FROM collection_pages WHERE id=?", (page_id,)).fetchone(): raise HTTPException(404, "Page not found") if not conn.execute("SELECT id FROM collection_pages WHERE id=?", (dep_id,)).fetchone(): raise HTTPException(404, "Dependency page not found") try: conn.execute("INSERT INTO page_dependencies (page_id, dependency_id, dependency_type) VALUES (?, ?, ?)", (page_id, dep_id, dtype)) conn.commit() except Exception as e: raise HTTPException(409, str(e)) from None audit_log(user, "page.add_dependency", "collection_page", page_id, str(dep_id), request) return {"status": "added"} @router.delete("/pages/{page_id}/dependencies/{dep_id}") def remove_dependency_v2(page_id: int, dep_id: int, request: Request, authorization: str | None = Header(default=None)): user = require_scope("write")(request, authorization) _v2_rate_check(request, user) with get_conn() as conn: conn.execute("DELETE FROM page_dependencies WHERE page_id=? AND dependency_id=?", (page_id, dep_id)) conn.commit() audit_log(user, "page.remove_dependency", "collection_page", page_id, str(dep_id), request) return {"status": "removed"} @router.get("/collections/{collection_id}/properties") def list_properties_v2(collection_id: int, request: Request, authorization: str | None = Header(default=None)): user = get_bearer_user(request, authorization) _v2_rate_check(request, user) with get_conn() as conn: if not conn.execute("SELECT id FROM collections WHERE id=?", (collection_id,)).fetchone(): raise HTTPException(404, "Collection not found") rows = conn.execute("SELECT * FROM collection_properties WHERE collection_id=? ORDER BY position", (collection_id,)).fetchall() return {"properties": [row_to_dict(r) for r in rows]}