"""Spreadsheet Canvas API Routes""" import logging from typing import Any, Dict, List, Optional from fastapi import Depends from pydantic import BaseModel from sqlalchemy.orm import Session from core.base_routes import BaseAPIRouter from core.canvas_sheets_service import SpreadsheetCanvasService from core.database import get_db logger = logging.getLogger(__name__) router = BaseAPIRouter(prefix="/api/canvas/sheets", tags=["canvas_sheets"]) class CreateSpreadsheetRequest(BaseModel): user_id: str title: str data: Dict[str, Any] canvas_id: Optional[str] = None agent_id: Optional[str] = None layout: str = "sheet" formulas: Optional[List[str]] = None class UpdateCellRequest(BaseModel): user_id: str cell_ref: str value: Any cell_type: str = "text" formula: Optional[str] = None class AddChartRequest(BaseModel): user_id: str chart_type: str data_range: str title: str = "" @router.post("/create") async def create_spreadsheet(request: CreateSpreadsheetRequest, db: Session = Depends(get_db)): """Create a new spreadsheet canvas.""" service = SpreadsheetCanvasService(db) result = service.create_spreadsheet_canvas( user_id=request.user_id, title=request.title, data=request.data, canvas_id=request.canvas_id, agent_id=request.agent_id, layout=request.layout, formulas=request.formulas ) if not result.get("success"): raise router.error_response( error_code="SPREADSHEET_CREATE_FAILED", message=result.get("error", "Failed to create spreadsheet canvas"), status_code=400 ) return result @router.put("/{canvas_id}/cell") async def update_cell(canvas_id: str, request: UpdateCellRequest, db: Session = Depends(get_db)): """Update a cell value.""" service = SpreadsheetCanvasService(db) result = service.update_cell( canvas_id=canvas_id, user_id=request.user_id, cell_ref=request.cell_ref, value=request.value, cell_type=request.cell_type, formula=request.formula ) if not result.get("success"): raise router.error_response( error_code="CELL_UPDATE_FAILED", message=result.get("error", "Failed to update cell"), status_code=400 ) return result @router.post("/{canvas_id}/chart") async def add_chart(canvas_id: str, request: AddChartRequest, db: Session = Depends(get_db)): """Add a chart to the spreadsheet.""" service = SpreadsheetCanvasService(db) result = service.add_chart( canvas_id=canvas_id, user_id=request.user_id, chart_type=request.chart_type, data_range=request.data_range, title=request.title ) if not result.get("success"): raise router.error_response( error_code="CHART_ADD_FAILED", message=result.get("error", "Failed to add chart"), status_code=400 ) return result @router.get("/{canvas_id}") async def get_spreadsheet(canvas_id: str, db: Session = Depends(get_db)): """Get a spreadsheet canvas.""" from sqlalchemy import desc from core.models import CanvasAudit audit = db.query(CanvasAudit).filter( CanvasAudit.canvas_id == canvas_id, CanvasAudit.canvas_type == "sheets" ).order_by(desc(CanvasAudit.created_at)).first() if not audit: raise router.not_found_error("Spreadsheet Canvas", canvas_id) return audit.audit_metadata