diff --git a/app/api/__init__.py b/app/api/__init__.py index ac916de2..f266a266 100644 --- a/app/api/__init__.py +++ b/app/api/__init__.py @@ -17,6 +17,7 @@ from app.api.google_drive import router as google_drive_router from app.api.logs import router as logs_router from app.api.onedrive import router as onedrive_router from app.api.openai import router as openai_router +from app.api.pipelines import router as pipelines_router from app.api.plans import router as plans_router from app.api.process import router as process_router from app.api.queue import router as queue_router @@ -60,3 +61,4 @@ router.include_router(webhooks_router) router.include_router(database_router) router.include_router(subscriptions_router) router.include_router(plans_router) +router.include_router(pipelines_router) diff --git a/app/api/files.py b/app/api/files.py index 87586664..ee5d2998 100644 --- a/app/api/files.py +++ b/app/api/files.py @@ -1559,3 +1559,75 @@ def assign_owner(request: Request, db: DbSession, owner_id: str = Query(...), fi "updated_count": updated, "owner_id": owner_id, } + + +# --------------------------------------------------------------------------- +# Pipeline assignment +# --------------------------------------------------------------------------- + + +@router.post("/files/{file_id}/assign-pipeline") +@require_login +def assign_pipeline_to_file( + request: Request, + file_id: int, + db: DbSession, + pipeline_id: int | None = None, +): + """Assign (or remove) a processing pipeline from a file. + + Path Parameters: + file_id: The file to update. + + Query / Body Parameters: + pipeline_id: The pipeline to assign. Pass ``null`` or omit to clear the + assignment (the system default will be used for future processing). + + Returns: + A summary dict with the file_id and updated pipeline_id. + + Raises: + HTTPException 404: If the file or pipeline does not exist / is not + accessible to the current user. + """ + from app.auth import get_current_user + from app.models import Pipeline + + user = get_current_user(request) + # Derive user identity the same way the pipelines API does (_get_user_id) + if user: + user_id: str = user.get("preferred_username") or user.get("email") or user.get("id") or "anonymous" + else: + user_id = "anonymous" + + is_admin_user = bool(user and user.get("is_admin")) + + file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first() + if not file_record: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found") + + # Non-admins may only update files they own (or unowned files in single-user mode) + owner_id = get_current_owner_id(request) + if not is_admin_user and file_record.owner_id is not None and file_record.owner_id != owner_id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found") + + if pipeline_id is not None: + pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first() + if not pipeline: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found") + # Check access: users can only assign their own pipelines or system pipelines (owner_id=None) + if not is_admin_user and pipeline.owner_id is not None and pipeline.owner_id != user_id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found") + + file_record.pipeline_id = pipeline_id + + try: + db.commit() + db.refresh(file_record) + except Exception as exc: + db.rollback() + logger.exception(f"Failed to assign pipeline to file id={file_id}: {exc}") + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to assign pipeline") + + logger.info(f"Pipeline {pipeline_id!r} assigned to file id={file_id}") + return {"file_id": file_id, "pipeline_id": file_record.pipeline_id} diff --git a/app/api/pipelines.py b/app/api/pipelines.py new file mode 100644 index 00000000..13182a80 --- /dev/null +++ b/app/api/pipelines.py @@ -0,0 +1,793 @@ +""" +Pipelines API endpoints. + +Provides full CRUD for processing pipelines and their steps. Pipelines are +user-specific: regular users can only manage their own pipelines, while admins +can also create and manage *system default* pipelines (owner_id = NULL) that +are visible to all users. + +Built-in step types are exposed via GET /api/pipelines/step-types so that UIs +can render the correct configuration form without hard-coding the catalogue. +""" + +import json +import logging +from typing import Annotated, Any + +from fastapi import APIRouter, Body, Depends, HTTPException, Request, status +from pydantic import BaseModel, Field +from sqlalchemy.orm import Session + +from app.auth import get_current_user, require_login +from app.database import get_db +from app.models import Pipeline, PipelineStep + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/pipelines", tags=["pipelines"]) + +DbSession = Annotated[Session, Depends(get_db)] + +# --------------------------------------------------------------------------- +# Built-in step type catalogue +# --------------------------------------------------------------------------- + +PIPELINE_STEP_TYPES: dict[str, dict[str, Any]] = { + "convert_to_pdf": { + "label": "Convert to PDF", + "description": "Convert non-PDF documents to PDF format using Gotenberg.", + "config_schema": {}, + }, + "check_duplicates": { + "label": "Check for Duplicates", + "description": "Compare file hash against existing documents to detect duplicates.", + "config_schema": {}, + }, + "ocr": { + "label": "OCR Processing", + "description": "Extract text using Azure Document Intelligence or local Tesseract.", + "config_schema": { + "force_cloud_ocr": { + "type": "boolean", + "default": False, + "description": "Always use cloud OCR even if the PDF already has embedded text.", + } + }, + }, + "extract_metadata": { + "label": "Metadata Extraction", + "description": "Extract structured metadata (document type, sender, recipient, tags) using AI.", + "config_schema": {}, + }, + "embed_metadata": { + "label": "Embed Metadata into PDF", + "description": "Write the extracted metadata into the PDF document properties.", + "config_schema": {}, + }, + "compute_embedding": { + "label": "Compute Text Embedding", + "description": "Compute semantic text embeddings for full-text and similarity search.", + "config_schema": {}, + }, + "send_to_destinations": { + "label": "Send to Storage Destinations", + "description": "Upload the processed document to all configured storage destinations.", + "config_schema": {}, + }, + "classify": { + "label": "Document Classification", + "description": "Classify the document type using AI without full metadata extraction.", + "config_schema": {}, + }, +} + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +MAX_STEPS_PER_PIPELINE = 50 +MAX_NAME_LENGTH = 255 + + +def _get_user_id(request: Request) -> str: + """Return a stable user identifier from the session.""" + user = get_current_user(request) + if user: + return user.get("preferred_username") or user.get("email") or user.get("id", "anonymous") + return "anonymous" + + +def _is_admin(request: Request) -> bool: + """Return True if the current session user is an admin.""" + user = get_current_user(request) + return bool(user and user.get("is_admin")) + + +def _can_access_pipeline(pipeline: Pipeline, user_id: str, admin: bool) -> bool: + """Return True if the user may read or write this pipeline.""" + # System pipelines (owner_id=NULL) are readable by everyone; only admins can write + if pipeline.owner_id is None: + return True + # Own pipeline + return pipeline.owner_id == user_id or admin + + +def _can_write_pipeline(pipeline: Pipeline, user_id: str, admin: bool) -> bool: + """Return True if the user may create/update/delete this pipeline.""" + if pipeline.owner_id is None: + return admin + return pipeline.owner_id == user_id or admin + + +def _serialize_step(step: PipelineStep) -> dict[str, Any]: + return { + "id": step.id, + "pipeline_id": step.pipeline_id, + "position": step.position, + "step_type": step.step_type, + "label": step.label, + "config": json.loads(step.config) if step.config else {}, + "enabled": step.enabled, + "created_at": step.created_at.isoformat() if step.created_at else None, + "updated_at": step.updated_at.isoformat() if step.updated_at else None, + } + + +def _serialize_pipeline(pipeline: Pipeline, include_steps: bool = False, db: Session | None = None) -> dict[str, Any]: + data: dict[str, Any] = { + "id": pipeline.id, + "owner_id": pipeline.owner_id, + "name": pipeline.name, + "description": pipeline.description, + "is_default": pipeline.is_default, + "is_active": pipeline.is_active, + "created_at": pipeline.created_at.isoformat() if pipeline.created_at else None, + "updated_at": pipeline.updated_at.isoformat() if pipeline.updated_at else None, + } + if include_steps and db is not None: + steps = ( + db.query(PipelineStep).filter(PipelineStep.pipeline_id == pipeline.id).order_by(PipelineStep.position).all() + ) + data["steps"] = [_serialize_step(s) for s in steps] + return data + + +# --------------------------------------------------------------------------- +# Pydantic schemas +# --------------------------------------------------------------------------- + + +class PipelineCreate(BaseModel): + """Body for creating a pipeline.""" + + name: str = Field(..., max_length=MAX_NAME_LENGTH, description="Human-readable pipeline name") + description: str | None = Field(default=None, max_length=4096) + is_default: bool = Field(default=False) + is_active: bool = Field(default=True) + + +class PipelineUpdate(BaseModel): + """Body for updating a pipeline (all fields optional).""" + + name: str | None = Field(default=None, max_length=MAX_NAME_LENGTH) + description: str | None = Field(default=None, max_length=4096) + is_default: bool | None = None + is_active: bool | None = None + + +class PipelineStepCreate(BaseModel): + """Body for adding a step to a pipeline.""" + + step_type: str = Field(..., description="One of the recognised step type keys") + label: str | None = Field(default=None, max_length=MAX_NAME_LENGTH) + config: dict[str, Any] = Field(default_factory=dict) + enabled: bool = Field(default=True) + position: int | None = Field(default=None, ge=0, description="Insertion position; appended at end if omitted") + + +class PipelineStepUpdate(BaseModel): + """Body for updating a pipeline step (all fields optional).""" + + step_type: str | None = None + label: str | None = Field(default=None, max_length=MAX_NAME_LENGTH) + config: dict[str, Any] | None = None + enabled: bool | None = None + position: int | None = Field(default=None, ge=0) + + +# --------------------------------------------------------------------------- +# Step-types catalogue endpoint (no auth required — it's public metadata) +# --------------------------------------------------------------------------- + + +@router.get("/step-types") +def list_step_types() -> dict[str, Any]: + """Return the catalogue of built-in pipeline step types. + + Returns: + A mapping of step_type key → metadata (label, description, config_schema). + """ + return PIPELINE_STEP_TYPES + + +# --------------------------------------------------------------------------- +# Pipeline CRUD +# --------------------------------------------------------------------------- + + +@router.get("") +@require_login +def list_pipelines(request: Request, db: DbSession) -> list[dict[str, Any]]: + """List pipelines visible to the current user. + + Regular users see: their own pipelines + system pipelines (owner_id=NULL). + Admins see: all pipelines from all users. + + Returns: + A list of pipeline objects (without steps — use GET /pipelines/{id} for steps). + """ + user_id = _get_user_id(request) + admin = _is_admin(request) + + if admin: + pipelines = db.query(Pipeline).order_by(Pipeline.owner_id.nullsfirst(), Pipeline.name).all() + else: + pipelines = ( + db.query(Pipeline) + .filter((Pipeline.owner_id == user_id) | (Pipeline.owner_id.is_(None))) + .order_by(Pipeline.owner_id.nullsfirst(), Pipeline.name) + .all() + ) + + return [_serialize_pipeline(p) for p in pipelines] + + +@router.post("", status_code=status.HTTP_201_CREATED) +@require_login +def create_pipeline(request: Request, db: DbSession, body: PipelineCreate) -> dict[str, Any]: + """Create a new pipeline for the current user. + + Admins can create system default pipelines by passing ``owner_id=null`` + via the body — however, that is handled implicitly: to create a system + pipeline, call ``POST /api/admin/pipelines`` (admin endpoint) instead. + Regular users always get their own user_id as owner. + + Returns: + The created pipeline object. + + Raises: + HTTPException 409: If a pipeline with the same name already exists for this owner. + """ + user_id = _get_user_id(request) + + name = body.name.strip() if body.name else "" + if not name: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="name is required", + ) + + # Enforce unique name per owner + existing = db.query(Pipeline).filter(Pipeline.owner_id == user_id, Pipeline.name == name).first() + if existing: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f"A pipeline named '{name}' already exists", + ) + + # If this pipeline is marked as default, unset the existing default for this user + if body.is_default: + _unset_default(db, user_id) + + pipeline = Pipeline( + owner_id=user_id, + name=name, + description=body.description, + is_default=body.is_default, + is_active=body.is_active, + ) + try: + db.add(pipeline) + db.commit() + db.refresh(pipeline) + except Exception as exc: + db.rollback() + logger.exception(f"Failed to create pipeline user={user_id}: {exc}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to create pipeline", + ) + + logger.info(f"Pipeline created: id={pipeline.id}, owner={user_id}, name={name!r}") + return _serialize_pipeline(pipeline) + + +@router.get("/{pipeline_id}") +@require_login +def get_pipeline(pipeline_id: int, request: Request, db: DbSession) -> dict[str, Any]: + """Return a single pipeline with its steps. + + Path Parameters: + pipeline_id: The ID of the pipeline. + + Returns: + The pipeline object including its ordered steps. + + Raises: + HTTPException 404: If the pipeline does not exist or is not accessible. + """ + user_id = _get_user_id(request) + admin = _is_admin(request) + + pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first() + if not pipeline or not _can_access_pipeline(pipeline, user_id, admin): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found") + + return _serialize_pipeline(pipeline, include_steps=True, db=db) + + +@router.put("/{pipeline_id}") +@require_login +def update_pipeline(pipeline_id: int, request: Request, db: DbSession, body: PipelineUpdate) -> dict[str, Any]: + """Update a pipeline's metadata. + + Path Parameters: + pipeline_id: The ID of the pipeline to update. + + Returns: + The updated pipeline object. + + Raises: + HTTPException 403: If the caller does not own this pipeline. + HTTPException 404: If the pipeline does not exist. + HTTPException 409: If the new name conflicts with an existing pipeline. + """ + user_id = _get_user_id(request) + admin = _is_admin(request) + + pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first() + if not pipeline or not _can_access_pipeline(pipeline, user_id, admin): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found") + + if not _can_write_pipeline(pipeline, user_id, admin): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot modify this pipeline") + + if body.name is not None: + new_name = body.name.strip() + if not new_name: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="name must not be empty", + ) + if new_name != pipeline.name: + conflict = ( + db.query(Pipeline) + .filter(Pipeline.owner_id == pipeline.owner_id, Pipeline.name == new_name, Pipeline.id != pipeline_id) + .first() + ) + if conflict: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f"A pipeline named '{new_name}' already exists", + ) + pipeline.name = new_name + + if body.description is not None: + pipeline.description = body.description + + if body.is_active is not None: + pipeline.is_active = body.is_active + + if body.is_default is not None: + if body.is_default and not pipeline.is_default: + _unset_default(db, pipeline.owner_id) + pipeline.is_default = body.is_default + + try: + db.commit() + db.refresh(pipeline) + except Exception as exc: + db.rollback() + logger.exception(f"Failed to update pipeline id={pipeline_id}: {exc}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to update pipeline", + ) + + logger.info(f"Pipeline updated: id={pipeline_id}, user={user_id}") + return _serialize_pipeline(pipeline, include_steps=True, db=db) + + +@router.delete("/{pipeline_id}", status_code=status.HTTP_204_NO_CONTENT) +@require_login +def delete_pipeline(pipeline_id: int, request: Request, db: DbSession) -> None: + """Delete a pipeline and all its steps. + + Path Parameters: + pipeline_id: The ID of the pipeline to delete. + + Raises: + HTTPException 403: If the caller does not own this pipeline. + HTTPException 404: If the pipeline does not exist. + """ + user_id = _get_user_id(request) + admin = _is_admin(request) + + pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first() + if not pipeline or not _can_access_pipeline(pipeline, user_id, admin): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found") + + if not _can_write_pipeline(pipeline, user_id, admin): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot delete this pipeline") + + try: + db.query(PipelineStep).filter(PipelineStep.pipeline_id == pipeline_id).delete() + db.delete(pipeline) + db.commit() + except Exception as exc: + db.rollback() + logger.exception(f"Failed to delete pipeline id={pipeline_id}: {exc}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to delete pipeline", + ) + + logger.info(f"Pipeline deleted: id={pipeline_id}, user={user_id}") + + +# --------------------------------------------------------------------------- +# Admin-only: create system (owner_id=NULL) pipeline +# --------------------------------------------------------------------------- + + +@router.post("/admin/system", status_code=status.HTTP_201_CREATED, tags=["admin-pipelines"]) +@require_login +def create_system_pipeline(request: Request, db: DbSession, body: PipelineCreate) -> dict[str, Any]: + """Create a system-level (owner_id=NULL) default pipeline. Admin only. + + System pipelines are visible to all users and can be set as the global + default. Only admins may create them. + + Returns: + The created system pipeline. + + Raises: + HTTPException 403: If the caller is not an admin. + HTTPException 409: If a system pipeline with the same name already exists. + """ + if not _is_admin(request): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") + + name = body.name.strip() if body.name else "" + if not name: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="name is required", + ) + + existing = db.query(Pipeline).filter(Pipeline.owner_id.is_(None), Pipeline.name == name).first() + if existing: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f"A system pipeline named '{name}' already exists", + ) + + if body.is_default: + _unset_default(db, None) + + pipeline = Pipeline( + owner_id=None, + name=name, + description=body.description, + is_default=body.is_default, + is_active=body.is_active, + ) + try: + db.add(pipeline) + db.commit() + db.refresh(pipeline) + except Exception as exc: + db.rollback() + logger.exception(f"Failed to create system pipeline: {exc}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to create system pipeline", + ) + + logger.info(f"System pipeline created: id={pipeline.id}, name={name!r}") + return _serialize_pipeline(pipeline) + + +# --------------------------------------------------------------------------- +# Step management +# --------------------------------------------------------------------------- + + +@router.post("/{pipeline_id}/steps", status_code=status.HTTP_201_CREATED) +@require_login +def add_step(pipeline_id: int, request: Request, db: DbSession, body: PipelineStepCreate) -> dict[str, Any]: + """Add a step to a pipeline. + + Steps are automatically appended at the end unless an explicit ``position`` + is supplied. All existing steps at or after the insertion position are + shifted forward by one. + + Path Parameters: + pipeline_id: The pipeline to add the step to. + + Returns: + The created step object. + + Raises: + HTTPException 403: If the caller cannot modify this pipeline. + HTTPException 404: If the pipeline does not exist. + HTTPException 422: If the step_type is not recognised. + HTTPException 409: If the maximum number of steps per pipeline is reached. + """ + user_id = _get_user_id(request) + admin = _is_admin(request) + + pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first() + if not pipeline or not _can_access_pipeline(pipeline, user_id, admin): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found") + + if not _can_write_pipeline(pipeline, user_id, admin): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot modify this pipeline") + + if body.step_type not in PIPELINE_STEP_TYPES: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"Unknown step type '{body.step_type}'. Valid types: {sorted(PIPELINE_STEP_TYPES)}", + ) + + current_count = db.query(PipelineStep).filter(PipelineStep.pipeline_id == pipeline_id).count() + if current_count >= MAX_STEPS_PER_PIPELINE: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f"Maximum of {MAX_STEPS_PER_PIPELINE} steps per pipeline reached", + ) + + # Determine insertion position + if body.position is None: + max_pos = ( + db.query(PipelineStep.position) + .filter(PipelineStep.pipeline_id == pipeline_id) + .order_by(PipelineStep.position.desc()) + .first() + ) + insert_pos = (max_pos[0] + 1) if max_pos else 0 + else: + insert_pos = body.position + # Shift existing steps + steps_to_shift = ( + db.query(PipelineStep) + .filter(PipelineStep.pipeline_id == pipeline_id, PipelineStep.position >= insert_pos) + .all() + ) + for s in steps_to_shift: + s.position += 1 + + step = PipelineStep( + pipeline_id=pipeline_id, + position=insert_pos, + step_type=body.step_type, + label=body.label, + config=json.dumps(body.config) if body.config else None, + enabled=body.enabled, + ) + try: + db.add(step) + db.commit() + db.refresh(step) + except Exception as exc: + db.rollback() + logger.exception(f"Failed to add step to pipeline id={pipeline_id}: {exc}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to add step", + ) + + logger.info(f"Step added: pipeline={pipeline_id}, step_type={body.step_type!r}, pos={insert_pos}") + return _serialize_step(step) + + +@router.put("/{pipeline_id}/steps/reorder", tags=["pipelines"]) +@require_login +def reorder_steps( + pipeline_id: int, + request: Request, + db: DbSession, + step_ids: list[int] = Body(..., description="Ordered list of step IDs representing the new order"), +) -> list[dict[str, Any]]: + """Replace the step order for a pipeline. + + Provide a complete ordered list of *all* step IDs. Their ``position`` + values will be reassigned 0, 1, 2, … in the given order. + + Path Parameters: + pipeline_id: The pipeline whose steps are being reordered. + + Returns: + The updated, ordered list of step objects. + + Raises: + HTTPException 422: If the provided list does not contain exactly the + current set of step IDs for this pipeline. + """ + user_id = _get_user_id(request) + admin = _is_admin(request) + + pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first() + if not pipeline or not _can_access_pipeline(pipeline, user_id, admin): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found") + + if not _can_write_pipeline(pipeline, user_id, admin): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot modify this pipeline") + + existing_steps = db.query(PipelineStep).filter(PipelineStep.pipeline_id == pipeline_id).all() + existing_ids = {s.id for s in existing_steps} + + if set(step_ids) != existing_ids or len(step_ids) != len(existing_ids): + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="step_ids must contain exactly the current step IDs for this pipeline", + ) + + step_map = {s.id: s for s in existing_steps} + for pos, sid in enumerate(step_ids): + step_map[sid].position = pos + + try: + db.commit() + except Exception as exc: + db.rollback() + logger.exception(f"Failed to reorder steps for pipeline id={pipeline_id}: {exc}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to reorder steps", + ) + + updated = ( + db.query(PipelineStep).filter(PipelineStep.pipeline_id == pipeline_id).order_by(PipelineStep.position).all() + ) + return [_serialize_step(s) for s in updated] + + +@router.put("/{pipeline_id}/steps/{step_id}") +@require_login +def update_step( + pipeline_id: int, step_id: int, request: Request, db: DbSession, body: PipelineStepUpdate +) -> dict[str, Any]: + """Update an existing pipeline step. + + Path Parameters: + pipeline_id: The owning pipeline. + step_id: The step to update. + + Returns: + The updated step object. + """ + user_id = _get_user_id(request) + admin = _is_admin(request) + + pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first() + if not pipeline or not _can_access_pipeline(pipeline, user_id, admin): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found") + + if not _can_write_pipeline(pipeline, user_id, admin): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot modify this pipeline") + + step = db.query(PipelineStep).filter(PipelineStep.id == step_id, PipelineStep.pipeline_id == pipeline_id).first() + if not step: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Step not found") + + if body.step_type is not None: + if body.step_type not in PIPELINE_STEP_TYPES: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"Unknown step type '{body.step_type}'", + ) + step.step_type = body.step_type + + if body.label is not None: + step.label = body.label + + if body.config is not None: + step.config = json.dumps(body.config) + + if body.enabled is not None: + step.enabled = body.enabled + + if body.position is not None and body.position != step.position: + old_pos = step.position + new_pos = body.position + if new_pos > old_pos: + # Moving down: shift intervening steps up + db.query(PipelineStep).filter( + PipelineStep.pipeline_id == pipeline_id, + PipelineStep.position > old_pos, + PipelineStep.position <= new_pos, + PipelineStep.id != step_id, + ).update({"position": PipelineStep.position - 1}) + else: + # Moving up: shift intervening steps down + db.query(PipelineStep).filter( + PipelineStep.pipeline_id == pipeline_id, + PipelineStep.position >= new_pos, + PipelineStep.position < old_pos, + PipelineStep.id != step_id, + ).update({"position": PipelineStep.position + 1}) + step.position = new_pos + + try: + db.commit() + db.refresh(step) + except Exception as exc: + db.rollback() + logger.exception(f"Failed to update step id={step_id}: {exc}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to update step", + ) + + return _serialize_step(step) + + +@router.delete("/{pipeline_id}/steps/{step_id}", status_code=status.HTTP_204_NO_CONTENT) +@require_login +def delete_step(pipeline_id: int, step_id: int, request: Request, db: DbSession) -> None: + """Delete a step from a pipeline. + + Path Parameters: + pipeline_id: The owning pipeline. + step_id: The step to delete. + """ + user_id = _get_user_id(request) + admin = _is_admin(request) + + pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first() + if not pipeline or not _can_access_pipeline(pipeline, user_id, admin): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found") + + if not _can_write_pipeline(pipeline, user_id, admin): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot modify this pipeline") + + step = db.query(PipelineStep).filter(PipelineStep.id == step_id, PipelineStep.pipeline_id == pipeline_id).first() + if not step: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Step not found") + + deleted_pos = step.position + try: + db.delete(step) + # Compact remaining step positions + db.query(PipelineStep).filter( + PipelineStep.pipeline_id == pipeline_id, + PipelineStep.position > deleted_pos, + ).update({"position": PipelineStep.position - 1}) + db.commit() + except Exception as exc: + db.rollback() + logger.exception(f"Failed to delete step id={step_id}: {exc}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to delete step", + ) + + logger.info(f"Step deleted: id={step_id}, pipeline={pipeline_id}") + + +# --------------------------------------------------------------------------- +# Helper: unset default flag for an owner +# --------------------------------------------------------------------------- + + +def _unset_default(db: Session, owner_id: str | None) -> None: + """Clear the is_default flag on all pipelines for the given owner.""" + if owner_id is None: + db.query(Pipeline).filter(Pipeline.owner_id.is_(None), Pipeline.is_default.is_(True)).update( + {"is_default": False} + ) + else: + db.query(Pipeline).filter(Pipeline.owner_id == owner_id, Pipeline.is_default.is_(True)).update( + {"is_default": False} + ) diff --git a/app/models.py b/app/models.py index b5857bb5..1287ff40 100644 --- a/app/models.py +++ b/app/models.py @@ -6,6 +6,7 @@ from app.database import Base # Foreign key constants _FILES_ID_FK = "files.id" +_PIPELINES_ID_FK = "pipelines.id" class DocumentMetadata(Base): @@ -79,6 +80,9 @@ class FileRecord(Base): # Pre-computed text embedding vector stored as JSON array of floats embedding = Column(Text, nullable=True) + # Processing pipeline assigned to this file (NULL = use system default) + pipeline_id = Column(Integer, ForeignKey(_PIPELINES_ID_FK), nullable=True, index=True) + # Timestamp when we inserted this record created_at = Column(DateTime(timezone=True), server_default=func.now(), index=True) @@ -259,3 +263,69 @@ class SubscriptionPlan(Base): created_at = Column(DateTime(timezone=True), server_default=func.now()) updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + + +class Pipeline(Base): + """User-defined processing pipeline: an ordered set of steps. + + Pipelines are user-specific. A pipeline with ``owner_id = NULL`` is a + *system default* pipeline that only admins may create. Regular users + create pipelines under their own ``owner_id``. When a file has no + explicit pipeline assigned, the active system default is used. + """ + + __tablename__ = "pipelines" + + id = Column(Integer, primary_key=True, index=True) + + # Owner of this pipeline. NULL = system/admin pipeline visible to everyone. + owner_id = Column(String, nullable=True, index=True) + + # Human-readable name (unique per owner) + name = Column(String(255), nullable=False) + + # Optional description + description = Column(Text, nullable=True) + + # When True this pipeline is the default for new files belonging to the owner + # (or the global default when owner_id is NULL). Only one pipeline per + # owner may be active default at a time — enforced at the application level. + is_default = Column(Boolean, nullable=False, default=False) + + # Soft-disable without deleting + is_active = Column(Boolean, nullable=False, default=True) + + created_at = Column(DateTime(timezone=True), server_default=func.now()) + updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + + +class PipelineStep(Base): + """A single step in a processing pipeline. + + Steps are executed in ascending ``position`` order. Each step has a + ``step_type`` that maps to a built-in processing action and an optional + ``config`` JSON blob with step-specific parameters. + """ + + __tablename__ = "pipeline_steps" + + id = Column(Integer, primary_key=True, index=True) + pipeline_id = Column(Integer, ForeignKey(_PIPELINES_ID_FK), nullable=False, index=True) + + # Execution order within the pipeline (lower = earlier) + position = Column(Integer, nullable=False, default=0) + + # One of the recognised step types (see PIPELINE_STEP_TYPES in pipelines.py) + step_type = Column(String(100), nullable=False) + + # Optional human-readable label override (defaults to step_type label) + label = Column(String(255), nullable=True) + + # JSON-encoded step-specific configuration dict + config = Column(Text, nullable=True) + + # When False this step is skipped during execution + enabled = Column(Boolean, nullable=False, default=True) + + created_at = Column(DateTime(timezone=True), server_default=func.now()) + updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) diff --git a/app/views/__init__.py b/app/views/__init__.py index 75eafbfb..309dd3d1 100644 --- a/app/views/__init__.py +++ b/app/views/__init__.py @@ -14,6 +14,7 @@ from app.views.general import router as general_router from app.views.google_drive import router as google_drive_router from app.views.license_routes import router as license_router # Add the license router from app.views.onedrive import router as onedrive_router +from app.views.pipelines import router as pipelines_router # Processing pipelines from app.views.plans import router as plans_router # Admin Plan Designer from app.views.queue import router as queue_router from app.views.search import router as search_router @@ -39,3 +40,4 @@ router.include_router(search_router) router.include_router(queue_router) router.include_router(subscriptions_router) # Pricing + subscription pages router.include_router(plans_router) # Admin Plan Designer +router.include_router(pipelines_router) # Processing pipelines diff --git a/app/views/pipelines.py b/app/views/pipelines.py new file mode 100644 index 00000000..a0619f7c --- /dev/null +++ b/app/views/pipelines.py @@ -0,0 +1,34 @@ +"""Pipelines view: management UI for processing pipelines.""" + +import logging + +from fastapi import HTTPException, Request, status + +from app.views.base import APIRouter, require_login, settings, templates + +logger = logging.getLogger(__name__) +router = APIRouter() + + +@router.get("/pipelines") +@require_login +async def pipelines_page(request: Request): + """Processing pipeline management page for the current user. + + Regular users manage their own pipelines. Admins additionally have access + to system-level pipelines through the same UI. + """ + try: + return templates.TemplateResponse( + "pipelines.html", + { + "request": request, + "app_version": settings.version, + }, + ) + except Exception as exc: + logger.error(f"Error loading pipelines page: {exc}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to load pipelines page", + ) diff --git a/frontend/templates/base.html b/frontend/templates/base.html index a6db72cb..49af2bad 100644 --- a/frontend/templates/base.html +++ b/frontend/templates/base.html @@ -63,6 +63,7 @@ Upload Files Search + Pipelines Pricing @@ -175,6 +176,7 @@ Upload Files Search + Pipelines Pricing diff --git a/frontend/templates/pipelines.html b/frontend/templates/pipelines.html new file mode 100644 index 00000000..fb36ec47 --- /dev/null +++ b/frontend/templates/pipelines.html @@ -0,0 +1,857 @@ +{% extends "base.html" %} +{% block title %}Processing Pipelines – DocuElevate{% endblock %} + +{% block content %} +
+ + +
+
+

+ + Processing Pipelines +

+

+ Define and manage custom document processing workflows. + System pipelines (created by admins) are shown with a + System + badge and are visible to all users. +

+
+ +
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
+
+

+

+
+ + +
+
+
+ +
+{% endblock %} + +{% block head_extra %} + +{% endblock %} diff --git a/migrations/versions/017_add_pipelines.py b/migrations/versions/017_add_pipelines.py new file mode 100644 index 00000000..d14abbf9 --- /dev/null +++ b/migrations/versions/017_add_pipelines.py @@ -0,0 +1,68 @@ +"""Add pipelines and pipeline_steps tables; add pipeline_id to files + +Revision ID: 017_add_pipelines +Revises: 016_add_userprofile_billing +Create Date: 2026-03-07 + +""" + +from typing import Union + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "017_add_pipelines" +down_revision: Union[str, None] = "016_add_userprofile_billing" +depends_on: Union[str, None] = None + + +def upgrade() -> None: + """Create pipelines, pipeline_steps tables and add pipeline_id FK to files.""" + op.create_table( + "pipelines", + sa.Column("id", sa.Integer(), primary_key=True, index=True), + sa.Column("owner_id", sa.String(), nullable=True, index=True), + sa.Column("name", sa.String(255), nullable=False), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("is_default", sa.Boolean(), nullable=False, server_default="0"), + sa.Column("is_active", sa.Boolean(), nullable=False, server_default="1"), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + + op.create_table( + "pipeline_steps", + sa.Column("id", sa.Integer(), primary_key=True, index=True), + sa.Column("pipeline_id", sa.Integer(), sa.ForeignKey("pipelines.id"), nullable=False, index=True), + sa.Column("position", sa.Integer(), nullable=False, server_default="0"), + sa.Column("step_type", sa.String(100), nullable=False), + sa.Column("label", sa.String(255), nullable=True), + sa.Column("config", sa.Text(), nullable=True), + sa.Column("enabled", sa.Boolean(), nullable=False, server_default="1"), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + + # Use batch mode for SQLite compatibility when altering the files table. + # SQLite does not support adding FK constraints inline via ALTER TABLE, so we + # add the plain integer column and define the FK reference at the model level. + with op.batch_alter_table("files") as batch_op: + batch_op.add_column( + sa.Column("pipeline_id", sa.Integer(), nullable=True), + ) + batch_op.create_foreign_key( + "fk_files_pipeline_id", + "pipelines", + ["pipeline_id"], + ["id"], + ) + + +def downgrade() -> None: + """Drop pipeline_id from files and remove pipeline tables.""" + with op.batch_alter_table("files") as batch_op: + batch_op.drop_column("pipeline_id") + + op.drop_table("pipeline_steps") + op.drop_table("pipelines") diff --git a/tests/conftest.py b/tests/conftest.py index 98cb5316..5f68bd9d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -62,6 +62,8 @@ from app.main import app as fastapi_app # noqa: E402 from app.models import ( # noqa: F401, E402 DocumentMetadata, FileRecord, + Pipeline, + PipelineStep, ProcessingLog, SavedSearch, UserProfile, diff --git a/tests/test_api_pipelines.py b/tests/test_api_pipelines.py new file mode 100644 index 00000000..ad7fcf33 --- /dev/null +++ b/tests/test_api_pipelines.py @@ -0,0 +1,467 @@ +"""Tests for the pipelines API endpoints. + +Covers CRUD operations for pipelines and steps, ownership/admin access control, +step reordering, and the assign-pipeline-to-file endpoint. +""" + +import json +from unittest.mock import patch + +import pytest + +from app.models import FileRecord, Pipeline, PipelineStep + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_file(db_session, owner_id=None): + """Insert a minimal FileRecord and return it. + + Default owner_id=None so tests work without an authenticated session. + """ + fr = FileRecord( + owner_id=owner_id, + filehash="abc123", + original_filename="test.pdf", + local_filename="/tmp/test.pdf", + file_size=1024, + mime_type="application/pdf", + ) + db_session.add(fr) + db_session.commit() + db_session.refresh(fr) + return fr + + +# --------------------------------------------------------------------------- +# Unit tests – step-types catalogue +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestStepTypesCatalogue: + """Tests for the step-types read-only catalogue endpoint.""" + + def test_step_types_returns_dict(self, client): + """GET /api/pipelines/step-types returns a dict of known types.""" + r = client.get("/api/pipelines/step-types") + assert r.status_code == 200 + data = r.json() + assert isinstance(data, dict) + # Core built-in types must be present + for key in ("convert_to_pdf", "ocr", "extract_metadata", "embed_metadata", "compute_embedding"): + assert key in data, f"Expected step type '{key}' in catalogue" + + def test_each_type_has_label_and_description(self, client): + """Each step-type entry has at least a label and description.""" + r = client.get("/api/pipelines/step-types") + for key, meta in r.json().items(): + assert "label" in meta, f"Step type '{key}' missing 'label'" + assert "description" in meta, f"Step type '{key}' missing 'description'" + + +# --------------------------------------------------------------------------- +# Integration tests – Pipeline CRUD +# --------------------------------------------------------------------------- + + +@pytest.mark.integration +class TestPipelineCRUD: + """Full CRUD test-suite for pipeline management.""" + + def test_list_pipelines_empty(self, client): + """List returns an empty array when no pipelines exist.""" + r = client.get("/api/pipelines") + assert r.status_code == 200 + assert r.json() == [] + + def test_create_pipeline(self, client): + """POST /api/pipelines creates a new pipeline.""" + r = client.post( + "/api/pipelines", + json={"name": "My Pipeline", "description": "Test description"}, + ) + assert r.status_code == 201 + data = r.json() + assert data["name"] == "My Pipeline" + assert data["description"] == "Test description" + assert data["is_default"] is False + assert data["is_active"] is True + assert data["id"] is not None + + def test_create_pipeline_duplicate_name_same_owner_rejected(self, client): + """Creating two pipelines with the same name is rejected with 409.""" + client.post("/api/pipelines", json={"name": "Dupe"}) + r = client.post("/api/pipelines", json={"name": "Dupe"}) + assert r.status_code == 409 + + def test_create_pipeline_empty_name_rejected(self, client): + """An empty pipeline name returns 422.""" + r = client.post("/api/pipelines", json={"name": " "}) + assert r.status_code == 422 + + def test_get_pipeline_includes_steps(self, client): + """GET /api/pipelines/{id} returns the pipeline with a steps array.""" + created = client.post("/api/pipelines", json={"name": "With Steps"}).json() + r = client.get(f"/api/pipelines/{created['id']}") + assert r.status_code == 200 + assert "steps" in r.json() + assert r.json()["steps"] == [] + + def test_get_pipeline_not_found(self, client): + """GET on a non-existent pipeline returns 404.""" + r = client.get("/api/pipelines/99999") + assert r.status_code == 404 + + def test_update_pipeline(self, client): + """PUT /api/pipelines/{id} updates name and description.""" + created = client.post("/api/pipelines", json={"name": "Original"}).json() + r = client.put( + f"/api/pipelines/{created['id']}", + json={"name": "Renamed", "description": "New desc"}, + ) + assert r.status_code == 200 + assert r.json()["name"] == "Renamed" + assert r.json()["description"] == "New desc" + + def test_update_pipeline_empty_name_rejected(self, client): + """Updating a pipeline with an empty name returns 422.""" + created = client.post("/api/pipelines", json={"name": "Good"}).json() + r = client.put(f"/api/pipelines/{created['id']}", json={"name": ""}) + assert r.status_code == 422 + + def test_update_pipeline_name_conflict_rejected(self, client): + """Updating a pipeline's name to one already taken returns 409.""" + client.post("/api/pipelines", json={"name": "Taken"}) + second = client.post("/api/pipelines", json={"name": "Other"}).json() + r = client.put(f"/api/pipelines/{second['id']}", json={"name": "Taken"}) + assert r.status_code == 409 + + def test_delete_pipeline(self, client): + """DELETE /api/pipelines/{id} removes the pipeline.""" + created = client.post("/api/pipelines", json={"name": "Deletable"}).json() + r = client.delete(f"/api/pipelines/{created['id']}") + assert r.status_code == 204 + assert client.get(f"/api/pipelines/{created['id']}").status_code == 404 + + def test_delete_pipeline_not_found(self, client): + """Deleting a non-existent pipeline returns 404.""" + r = client.delete("/api/pipelines/99999") + assert r.status_code == 404 + + def test_is_default_flag(self, client): + """Setting is_default=True marks the pipeline as default.""" + r = client.post("/api/pipelines", json={"name": "Default Pipeline", "is_default": True}) + assert r.status_code == 201 + assert r.json()["is_default"] is True + + def test_only_one_default_per_owner(self, client): + """When a new default is set, the old one is cleared.""" + first = client.post("/api/pipelines", json={"name": "First Default", "is_default": True}).json() + second = client.post("/api/pipelines", json={"name": "Second Default", "is_default": True}).json() + + assert second["is_default"] is True + # The first should no longer be default + first_updated = client.get(f"/api/pipelines/{first['id']}").json() + assert first_updated["is_default"] is False + + def test_list_returns_created_pipeline(self, client): + """After creating a pipeline it appears in the list.""" + client.post("/api/pipelines", json={"name": "Visible"}) + r = client.get("/api/pipelines") + names = [p["name"] for p in r.json()] + assert "Visible" in names + + +# --------------------------------------------------------------------------- +# Integration tests – Step management +# --------------------------------------------------------------------------- + + +@pytest.mark.integration +class TestPipelineStepManagement: + """Tests for adding, updating, deleting, and reordering pipeline steps.""" + + def _create_pipeline(self, client, name="Test Pipeline"): + return client.post("/api/pipelines", json={"name": name}).json() + + def test_add_step(self, client): + """POST /api/pipelines/{id}/steps adds a step.""" + p = self._create_pipeline(client) + r = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "ocr"}) + assert r.status_code == 201 + step = r.json() + assert step["step_type"] == "ocr" + assert step["position"] == 0 + assert step["enabled"] is True + + def test_add_step_with_config(self, client): + """A step can be added with a custom config dict.""" + p = self._create_pipeline(client) + r = client.post( + f"/api/pipelines/{p['id']}/steps", + json={"step_type": "ocr", "config": {"force_cloud_ocr": True}}, + ) + assert r.status_code == 201 + assert r.json()["config"]["force_cloud_ocr"] is True + + def test_add_step_with_custom_label(self, client): + """A step can override the default label.""" + p = self._create_pipeline(client) + r = client.post( + f"/api/pipelines/{p['id']}/steps", + json={"step_type": "convert_to_pdf", "label": "My Converter"}, + ) + assert r.status_code == 201 + assert r.json()["label"] == "My Converter" + + def test_add_invalid_step_type_rejected(self, client): + """An unrecognised step type returns 422.""" + p = self._create_pipeline(client) + r = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "nonexistent_step"}) + assert r.status_code == 422 + + def test_steps_appended_in_order(self, client): + """Multiple steps are appended in position order.""" + p = self._create_pipeline(client) + client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "convert_to_pdf"}) + client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "ocr"}) + client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "extract_metadata"}) + + details = client.get(f"/api/pipelines/{p['id']}").json() + types = [s["step_type"] for s in details["steps"]] + assert types == ["convert_to_pdf", "ocr", "extract_metadata"] + + def test_update_step(self, client): + """PUT /api/pipelines/{id}/steps/{step_id} updates enabled flag.""" + p = self._create_pipeline(client) + step = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "ocr"}).json() + + r = client.put( + f"/api/pipelines/{p['id']}/steps/{step['id']}", + json={"enabled": False}, + ) + assert r.status_code == 200 + assert r.json()["enabled"] is False + + def test_update_step_not_found(self, client): + """Updating a step on the wrong pipeline returns 404.""" + p = self._create_pipeline(client) + r = client.put(f"/api/pipelines/{p['id']}/steps/99999", json={"enabled": False}) + assert r.status_code == 404 + + def test_delete_step(self, client): + """DELETE /api/pipelines/{id}/steps/{step_id} removes the step.""" + p = self._create_pipeline(client) + step = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "ocr"}).json() + + r = client.delete(f"/api/pipelines/{p['id']}/steps/{step['id']}") + assert r.status_code == 204 + + details = client.get(f"/api/pipelines/{p['id']}").json() + assert details["steps"] == [] + + def test_delete_step_compacts_positions(self, client): + """After deleting a step, remaining steps have contiguous positions.""" + p = self._create_pipeline(client) + s1 = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "convert_to_pdf"}).json() + client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "ocr"}) + client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "extract_metadata"}) + + client.delete(f"/api/pipelines/{p['id']}/steps/{s1['id']}") + + details = client.get(f"/api/pipelines/{p['id']}").json() + positions = [s["position"] for s in details["steps"]] + assert positions == sorted(positions) + assert positions[0] == 0 + + def test_reorder_steps(self, client): + """PUT /api/pipelines/{id}/steps/reorder reorders all steps.""" + p = self._create_pipeline(client) + s1 = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "convert_to_pdf"}).json() + s2 = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "ocr"}).json() + s3 = client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "extract_metadata"}).json() + + # Reverse order + r = client.put( + f"/api/pipelines/{p['id']}/steps/reorder", + json=[s3["id"], s2["id"], s1["id"]], + ) + assert r.status_code == 200 + types = [s["step_type"] for s in r.json()] + assert types == ["extract_metadata", "ocr", "convert_to_pdf"] + + def test_reorder_steps_invalid_ids_rejected(self, client): + """Providing wrong step IDs returns 422.""" + p = self._create_pipeline(client) + client.post(f"/api/pipelines/{p['id']}/steps", json={"step_type": "ocr"}) + + r = client.put(f"/api/pipelines/{p['id']}/steps/reorder", json=[99999]) + assert r.status_code == 422 + + +# --------------------------------------------------------------------------- +# Integration tests – System pipeline (admin endpoint) +# --------------------------------------------------------------------------- + + +@pytest.mark.integration +class TestSystemPipeline: + """Tests for the admin system-pipeline creation endpoint.""" + + def test_create_system_pipeline_as_admin(self, client): + """An admin can create a system pipeline (owner_id=NULL).""" + with patch("app.api.pipelines._is_admin", return_value=True): + r = client.post( + "/api/pipelines/admin/system", + json={"name": "Global Default", "is_default": True}, + ) + assert r.status_code == 201 + data = r.json() + assert data["owner_id"] is None + assert data["is_default"] is True + + def test_create_system_pipeline_as_non_admin_forbidden(self, client): + """A non-admin user cannot create a system pipeline.""" + r = client.post( + "/api/pipelines/admin/system", + json={"name": "Should Fail"}, + ) + assert r.status_code == 403 + + +# --------------------------------------------------------------------------- +# Integration tests – File pipeline assignment +# --------------------------------------------------------------------------- + + +@pytest.mark.integration +class TestAssignPipelineToFile: + """Tests for POST /api/files/{id}/assign-pipeline.""" + + def test_assign_pipeline_to_file(self, client, db_session): + """Assigning a pipeline to a file stores pipeline_id on the record.""" + # File with no owner so anonymous test session can access it + fr = _make_file(db_session, owner_id=None) + pipeline = client.post("/api/pipelines", json={"name": "Assign Test"}).json() + + r = client.post(f"/api/files/{fr.id}/assign-pipeline?pipeline_id={pipeline['id']}") + assert r.status_code == 200 + data = r.json() + assert data["file_id"] == fr.id + assert data["pipeline_id"] == pipeline["id"] + + db_session.refresh(fr) + assert fr.pipeline_id == pipeline["id"] + + def test_clear_pipeline_from_file(self, client, db_session): + """Passing no pipeline_id clears the assignment.""" + fr = _make_file(db_session, owner_id=None) + pipeline = client.post("/api/pipelines", json={"name": "Clearable"}).json() + client.post(f"/api/files/{fr.id}/assign-pipeline?pipeline_id={pipeline['id']}") + + r = client.post(f"/api/files/{fr.id}/assign-pipeline") + assert r.status_code == 200 + assert r.json()["pipeline_id"] is None + + def test_assign_nonexistent_pipeline_returns_404(self, client, db_session): + """Assigning a non-existent pipeline returns 404.""" + fr = _make_file(db_session, owner_id=None) + r = client.post(f"/api/files/{fr.id}/assign-pipeline?pipeline_id=99999") + assert r.status_code == 404 + + def test_assign_pipeline_to_nonexistent_file_returns_404(self, client): + """Assigning a pipeline to a non-existent file returns 404.""" + r = client.post("/api/files/99999/assign-pipeline?pipeline_id=1") + assert r.status_code == 404 + + +# --------------------------------------------------------------------------- +# Unit tests – API helper logic +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestPipelineAPIHelpers: + """Unit tests for API helper functions (no DB required).""" + + def test_serialize_step_returns_expected_keys(self, db_session): + """_serialize_step returns all required fields.""" + from app.api.pipelines import _serialize_step + + # Build a minimal in-memory step + p = Pipeline(owner_id="u1", name="P", is_default=False, is_active=True) + db_session.add(p) + db_session.commit() + + s = PipelineStep( + pipeline_id=p.id, + position=0, + step_type="ocr", + label="OCR", + config=json.dumps({"force_cloud_ocr": False}), + enabled=True, + ) + db_session.add(s) + db_session.commit() + + result = _serialize_step(s) + for key in ("id", "pipeline_id", "position", "step_type", "label", "config", "enabled"): + assert key in result, f"Missing key '{key}' in serialized step" + assert result["config"] == {"force_cloud_ocr": False} + + def test_serialize_pipeline_returns_expected_keys(self, db_session): + """_serialize_pipeline returns all required fields.""" + from app.api.pipelines import _serialize_pipeline + + p = Pipeline(owner_id="u1", name="MyPipeline", is_default=True, is_active=True) + db_session.add(p) + db_session.commit() + + result = _serialize_pipeline(p) + for key in ("id", "owner_id", "name", "description", "is_default", "is_active"): + assert key in result, f"Missing key '{key}' in serialized pipeline" + + def test_can_access_system_pipeline(self): + """Anyone can read a system pipeline (owner_id=None).""" + from app.api.pipelines import _can_access_pipeline + + p = Pipeline(owner_id=None, name="System", is_default=False, is_active=True) + assert _can_access_pipeline(p, "any_user", admin=False) is True + + def test_cannot_write_system_pipeline_as_regular_user(self): + """Regular users cannot modify system pipelines.""" + from app.api.pipelines import _can_write_pipeline + + p = Pipeline(owner_id=None, name="System", is_default=False, is_active=True) + assert _can_write_pipeline(p, "regular_user", admin=False) is False + + def test_admin_can_write_system_pipeline(self): + """Admins can modify system pipelines.""" + from app.api.pipelines import _can_write_pipeline + + p = Pipeline(owner_id=None, name="System", is_default=False, is_active=True) + assert _can_write_pipeline(p, "admin", admin=True) is True + + def test_user_can_access_own_pipeline(self): + """A user can access pipelines they own.""" + from app.api.pipelines import _can_access_pipeline + + p = Pipeline(owner_id="user1", name="Mine", is_default=False, is_active=True) + assert _can_access_pipeline(p, "user1", admin=False) is True + + def test_user_cannot_access_other_users_pipeline(self): + """A regular user cannot access another user's pipeline.""" + from app.api.pipelines import _can_access_pipeline + + p = Pipeline(owner_id="user1", name="Theirs", is_default=False, is_active=True) + assert _can_access_pipeline(p, "user2", admin=False) is False + + def test_admin_can_access_any_pipeline(self): + """Admins can access any pipeline regardless of owner.""" + from app.api.pipelines import _can_access_pipeline + + p = Pipeline(owner_id="user99", name="Private", is_default=False, is_active=True) + assert _can_access_pipeline(p, "admin", admin=True) is True