| """Batch upload API.""" |
|
|
| from __future__ import annotations |
|
|
| import uuid |
| from pathlib import Path |
|
|
| from fastapi import APIRouter, Depends, File, HTTPException, UploadFile |
| from sqlalchemy import select |
| from sqlalchemy.ext.asyncio import AsyncSession |
|
|
| from app.api.serializers import batch_out, document_out |
| from app.db import get_session |
| from app.models import Batch, Document |
| from app.schemas import BatchOut, DocStatus, DocumentOut |
| from app.services.ingest import ensure_upload_dir, save_upload |
| from app.services.paths import resolve_stored_path, to_stored_path |
| from app.services.queue import enqueue_batch |
|
|
| router = APIRouter(prefix="/api/batches", tags=["batches"]) |
|
|
|
|
| @router.post("", response_model=BatchOut) |
| async def create_batch( |
| files: list[UploadFile] = File(...), |
| session: AsyncSession = Depends(get_session), |
| ) -> BatchOut: |
| """Upload one or more files / ZIPs as a batch.""" |
| if not files: |
| raise HTTPException(400, "No files uploaded") |
|
|
| batch_id = str(uuid.uuid4()) |
| root = ensure_upload_dir() |
| batch_dir = root / batch_id |
| batch_dir.mkdir(parents=True, exist_ok=True) |
|
|
| saved: list[tuple[str, Path, str]] = [] |
| errors: list[str] = [] |
| for upload in files: |
| try: |
| saved.extend(await save_upload(upload, batch_dir)) |
| except ValueError as exc: |
| errors.append(str(exc)) |
|
|
| if not saved: |
| raise HTTPException( |
| 400, |
| "; ".join(errors) or "No valid files", |
| ) |
|
|
| batch = Batch( |
| id=batch_id, |
| status="pending", |
| total=len(saved), |
| processed=0, |
| failed=0, |
| ) |
| session.add(batch) |
|
|
| for name, path, mime in saved: |
| session.add( |
| Document( |
| id=str(uuid.uuid4()), |
| batch_id=batch_id, |
| filename=name, |
| file_path=to_stored_path(path), |
| mime_type=mime, |
| status=DocStatus.QUEUED.value, |
| ) |
| ) |
| await session.commit() |
| await session.refresh(batch) |
| enqueue_batch(batch_id) |
| return batch_out(batch) |
|
|
|
|
| @router.get("/{batch_id}", response_model=BatchOut) |
| async def get_batch( |
| batch_id: str, |
| session: AsyncSession = Depends(get_session), |
| ) -> BatchOut: |
| """Batch status and progress.""" |
| batch = await session.get(Batch, batch_id) |
| if not batch: |
| raise HTTPException(404, "Batch not found") |
| return batch_out(batch) |
|
|
|
|
| @router.get( |
| "/{batch_id}/documents", |
| response_model=list[DocumentOut], |
| ) |
| async def list_batch_documents( |
| batch_id: str, |
| session: AsyncSession = Depends(get_session), |
| ) -> list[DocumentOut]: |
| """Documents belonging to a batch.""" |
| batch = await session.get(Batch, batch_id) |
| if not batch: |
| raise HTTPException(404, "Batch not found") |
| result = await session.execute( |
| select(Document).where(Document.batch_id == batch_id) |
| ) |
| out = [] |
| for d in result.scalars().all(): |
| try: |
| if resolve_stored_path(d.file_path).is_file(): |
| out.append(document_out(d)) |
| except FileNotFoundError: |
| continue |
| return out |
|
|
