"""v2 document routes — unstructured files (pdf / docx / txt) only. UNSTRUCTURED_V2_PLAN U3. The FE calls these instead of Go's `/api/v1/document*` for unstructured types; tabular files (csv / xlsx) stay entirely on Go, and so does **delete** for every type (U-D7 — Go's delete is authenticated and already removes the original and every embedding by `document_id`, v2 rows included). Request/response shapes, messages and status codes MIRROR Go's handlers (`Orchestrator-Agent-Service/internal/documents/handler.go`, read 2026-09-23) so the FE changes only the URL prefix. Envelope: `{status, message, data}`. ⚠️ Unlike Go, these routes do not verify a bearer token — the whole Python surface is unauthenticated by design until DEV_PLAN #43. Recorded under U-D7. """ # No `from __future__ import annotations` here: FastAPI must resolve the # `UploadFile | None` parameter annotation at import time. import uuid from pathlib import Path from types import SimpleNamespace from typing import Any from fastapi import APIRouter, BackgroundTasks, Depends, File, Form, Request, UploadFile from fastapi.responses import JSONResponse from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from src.db.postgres.connection import get_db from src.db.postgres.models import Document from src.document.image_token import resolve_image_urls from src.document.ingest_v2 import UNSTRUCTURED_TYPES, document_ingest_v2 from src.middlewares.logging import get_logger from src.middlewares.rate_limit import limiter logger = get_logger("documents_v2_api") router = APIRouter(prefix="/api/v2", tags=["Documents v2"]) MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024 # Go's maxFileSizeBytes _CONTENT_TYPES = { "pdf": "application/pdf", "docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", "txt": "text/plain; charset=utf-8", } def _envelope(code: int, status: str, message: str, data: Any = None) -> JSONResponse: body: dict[str, Any] = {"status": status, "message": message} if data is not None: body["data"] = data return JSONResponse(status_code=code, content=body) def _error(code: int, message: str) -> JSONResponse: return _envelope(code, "error", message) def _doc_json(d: Document) -> dict[str, Any]: """Go's Document model. `processed_at` / `error_message` are omitempty there.""" out: dict[str, Any] = { "id": d.id, "user_id": d.user_id, "filename": d.filename, "blob_name": d.blob_name, "file_size": d.file_size or 0, "file_type": d.file_type or "", "status": d.status or "", "chunks_count": getattr(d, "chunks_count", None) or 0, "created_at": d.created_at.isoformat() if d.created_at else None, } if d.processed_at: out["processed_at"] = d.processed_at.isoformat() if d.error_message: out["error_message"] = d.error_message return out def _extension(filename: str) -> str: return Path(filename or "").suffix.lstrip(".").lower() @router.get("/documents/doctypes", summary="Supported unstructured document types (v2)") async def list_doctypes(): data = [ {"type": t, "max_size_mb": 10, "status": "active", "message": None} for t in UNSTRUCTURED_TYPES ] return _envelope(200, "success", "supported document types", data) @router.get("/documents/{user_id}", summary="List the user's documents") async def list_documents(user_id: str, db: AsyncSession = Depends(get_db)): """Every document the user owns (all types — same table and order as Go's list).""" try: rows = ( await db.execute( select(Document) .where(Document.user_id == user_id) .order_by(Document.created_at.desc()) ) ).scalars().all() except Exception as e: logger.error("list documents failed", user_id=user_id, error=repr(e)) return _error(500, "internal server error") return _envelope(200, "success", "documents", [_doc_json(d) for d in rows]) @router.post("/document/upload", summary="Upload a pdf / docx / txt (max 10 MB)") @limiter.limit("10/minute") # Go: WithRateLimit(10, …) on this route async def upload_document( request: Request, user_id: str = Form(""), file: UploadFile | None = File(None), db: AsyncSession = Depends(get_db), ): if not user_id.strip(): return _error(400, "user_id is required") if file is None: return _error(400, "file is required") ext = _extension(file.filename or "") if ext not in UNSTRUCTURED_TYPES: return _error(400, "unsupported file type") content = await file.read(MAX_FILE_SIZE_BYTES + 1) if len(content) > MAX_FILE_SIZE_BYTES: return _error(400, "file too large (max 10 MB)") from src.storage.object_storage.supabase_s3 import object_storage doc_id = str(uuid.uuid4()) doc = Document( id=doc_id, user_id=user_id, filename=file.filename, blob_name=f"{user_id}/{doc_id}.{ext}", # Go's exact blob_name format file_size=len(content), file_type=ext, status="uploading", ) try: db.add(doc) await db.commit() await db.refresh(doc) except Exception as e: logger.error("upload: db record failed", user_id=user_id, error=repr(e)) return _error(500, "upload failed") try: await object_storage.upload_bytes(content, doc.blob_name, _CONTENT_TYPES[ext]) except Exception as e: doc.status, doc.error_message = "failed", repr(e)[:500] await db.commit() return _error(500, "upload failed") doc.status = "uploaded" await db.commit() await db.refresh(doc) logger.info("upload: completed", document_id=doc_id, user_id=user_id, size=len(content)) return _envelope(201, "success", "file uploaded", _doc_json(doc)) def _json_body(properties: dict[str, Any], example: dict[str, Any]) -> dict[str, Any]: """OpenAPI-only body description (added 2026-09-24). These routes read `request.json()` themselves so their errors stay Go-shaped (400 `invalid request body`, not FastAPI's 422). The side effect: FastAPI declared no request body, so Swagger's "Try it out" showed no editor and POSTed nothing — every Swagger test answered 400. This documents the body; validation is unchanged. """ return {"requestBody": {"required": True, "content": {"application/json": { "schema": {"type": "object", "required": list(properties), "properties": properties}, "example": example, }}}} @router.post( "/document/process", summary="Parse + embed an uploaded document (async)", openapi_extra=_json_body( {"document_id": {"type": "string"}, "user_id": {"type": "string"}}, {"document_id": "", "user_id": ""}, ), ) async def process_document( request: Request, background: BackgroundTasks, db: AsyncSession = Depends(get_db), ): try: body = await request.json() document_id = str(body.get("document_id") or "") user_id = str(body.get("user_id") or "") except Exception: return _error(400, "invalid request body") if not document_id or not user_id: return _error(400, "document_id and user_id are required") doc = await db.get(Document, document_id) if doc is None: return _error(404, "document not found") if doc.user_id != user_id: return _error(403, "access denied") if (doc.file_type or "").lower() not in UNSTRUCTURED_TYPES: # Tabular processing (parquet + catalog) is Go's; never half-do it here. return _error(400, "unsupported file type") doc.status, doc.error_message = "processing", None await db.commit() snapshot = SimpleNamespace( id=doc.id, user_id=doc.user_id, filename=doc.filename, file_type=(doc.file_type or "").lower(), blob_name=doc.blob_name, ) background.add_task(document_ingest_v2.process, snapshot) return _envelope( 202, "success", "document processing started", {"document_id": doc.id, "file_type": snapshot.file_type, "status": "processing"}, ) @router.post( "/document/image-token", summary="Short-lived URLs for figure markers", openapi_extra=_json_body( { "user_id": {"type": "string"}, "analysis_id": {"type": "string"}, "asset_ids": {"type": "array", "items": {"type": "string"}, "maxItems": 50}, }, {"user_id": "", "analysis_id": "", "asset_ids": ["4f72f8b170d1787b"]}, ), ) async def image_token(request: Request): """Resolve `asset://` markers to 1-hour signed URLs (U4). See image_token.py.""" try: body = await request.json() user_id = str(body.get("user_id") or "") analysis_id = str(body.get("analysis_id") or "") asset_ids = body.get("asset_ids") except Exception: return _error(400, "invalid request body") if not user_id or not analysis_id or not isinstance(asset_ids, list): return _error(400, "user_id, analysis_id and asset_ids are required") if len(asset_ids) > 50: return _error(400, "at most 50 asset_ids per request") try: urls, expires_at = await resolve_image_urls(user_id, analysis_id, asset_ids) except Exception as e: logger.error("image-token failed", user_id=user_id, error=repr(e)) return _error(500, "internal server error") return _envelope(200, "success", "image urls", {"urls": urls, "expires_at": expires_at})