gh-christianlouis-docuelevate/app/api/imap_profiles.py

"""API endpoints for managing IMAP ingestion profiles.

Ingestion profiles allow fine-grained control over which attachment types are
accepted when ingesting emails via IMAP.  Each profile carries a list of enabled
file-type categories (e.g. ``["pdf", "office", "images"]``) drawn from the
canonical set defined in :mod:`app.utils.allowed_types`.

Built-in system profiles (``is_builtin=True``) are read-only and cannot be
deleted or modified.  Users may create their own profiles which are private to
their ``owner_id``.  System-level global profiles (``owner_id=None``) are visible
to all users but can only be created by administrators.
"""

import json
import logging
from datetime import datetime, timezone
from typing import Annotated, Any

from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session

from app.database import get_db
from app.models import ImapIngestionProfile
from app.utils.allowed_types import FILE_TYPE_CATEGORIES
from app.utils.user_scope import get_current_owner_id

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/imap-profiles", tags=["imap-profiles"])

DbSession = Annotated[Session, Depends(get_db)]

# ---------------------------------------------------------------------------
# Auth helpers
# ---------------------------------------------------------------------------


def _get_owner_id(request: Request) -> str:
    """Return the current user's owner ID, raising 401 if unauthenticated."""
    owner_id = get_current_owner_id(request)
    if owner_id is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
    return owner_id


CurrentOwner = Annotated[str, Depends(_get_owner_id)]

# ---------------------------------------------------------------------------
# Pydantic schemas
# ---------------------------------------------------------------------------

_VALID_CATEGORIES = set(FILE_TYPE_CATEGORIES.keys())


class ImapProfileCreate(BaseModel):
    """Schema for creating a new ingestion profile."""

    name: str = Field(..., min_length=1, max_length=255, description="Human-readable profile name")
    description: str | None = Field(default=None, description="Optional description")
    allowed_categories: list[str] = Field(
        ...,
        min_length=1,
        description=(f"List of enabled file-type category keys.  Valid values: {sorted(_VALID_CATEGORIES)}"),
    )


class ImapProfileUpdate(BaseModel):
    """Schema for updating an existing profile (all fields optional)."""

    name: str | None = Field(default=None, min_length=1, max_length=255)
    description: str | None = None
    allowed_categories: list[str] | None = Field(default=None, min_length=1)


# ---------------------------------------------------------------------------
# Validation helpers
# ---------------------------------------------------------------------------


def _validate_categories(categories: list[str]) -> list[str]:
    """Raise 422 if any category key is unknown; return the cleaned list."""
    unknown = [c for c in categories if c not in _VALID_CATEGORIES]
    if unknown:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"Unknown category key(s): {unknown}. Valid keys: {sorted(_VALID_CATEGORIES)}",
        )
    # Deduplicate while preserving order
    seen: set[str] = set()
    result: list[str] = []
    for cat in categories:
        if cat not in seen:
            seen.add(cat)
            result.append(cat)
    return result


# ---------------------------------------------------------------------------
# Serialisation
# ---------------------------------------------------------------------------


def _to_response(profile: ImapIngestionProfile) -> dict[str, Any]:
    """Serialize a profile row to a response dict."""
    try:
        categories = json.loads(profile.allowed_categories)
    except (ValueError, TypeError):
        categories = []

    # Enrich categories with display metadata
    categories_detail = [
        {
            "key": cat,
            "label": FILE_TYPE_CATEGORIES[cat]["label"] if cat in FILE_TYPE_CATEGORIES else cat,
            "description": FILE_TYPE_CATEGORIES[cat]["description"] if cat in FILE_TYPE_CATEGORIES else "",
        }
        for cat in categories
    ]

    return {
        "id": profile.id,
        "name": profile.name,
        "description": profile.description,
        "owner_id": profile.owner_id,
        "allowed_categories": categories,
        "categories_detail": categories_detail,
        "is_builtin": profile.is_builtin,
        "created_at": profile.created_at.isoformat() if profile.created_at else None,
        "updated_at": profile.updated_at.isoformat() if profile.updated_at else None,
    }


# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------


@router.get("/categories", summary="List available file-type categories")
def list_categories(request: Request, owner_id: CurrentOwner) -> list[dict[str, Any]]:
    """Return the full list of file-type categories that can be used in profiles."""
    return [
        {
            "key": key,
            "label": info["label"],
            "description": info["description"],
        }
        for key, info in FILE_TYPE_CATEGORIES.items()
    ]


@router.get("/", summary="List ingestion profiles visible to the current user")
def list_profiles(request: Request, db: DbSession, owner_id: CurrentOwner) -> list[dict[str, Any]]:
    """Return all profiles: system-global (owner_id=NULL) and the user's own profiles."""
    profiles = (
        db.query(ImapIngestionProfile)
        .filter(
            # SQLAlchemy requires `== None` for IS NULL comparison in ORM filters
            (ImapIngestionProfile.owner_id == None) | (ImapIngestionProfile.owner_id == owner_id)  # noqa: E711
        )
        .order_by(ImapIngestionProfile.is_builtin.desc(), ImapIngestionProfile.id)
        .all()
    )
    return [_to_response(p) for p in profiles]


@router.post("/", status_code=status.HTTP_201_CREATED, summary="Create a new ingestion profile")
def create_profile(request: Request, body: ImapProfileCreate, db: DbSession, owner_id: CurrentOwner) -> dict[str, Any]:
    """Create a new ingestion profile owned by the current user."""
    categories = _validate_categories(body.allowed_categories)

    profile = ImapIngestionProfile(
        name=body.name,
        description=body.description,
        owner_id=owner_id,
        allowed_categories=json.dumps(categories),
        is_builtin=False,
    )
    try:
        db.add(profile)
        db.commit()
        db.refresh(profile)
    except Exception:
        db.rollback()
        raise

    logger.info("User %s created IMAP ingestion profile %d ('%s')", owner_id, profile.id, body.name)
    return _to_response(profile)


@router.get("/{profile_id}", summary="Get a single ingestion profile")
def get_profile(profile_id: int, request: Request, db: DbSession, owner_id: CurrentOwner) -> dict[str, Any]:
    """Return a single profile by ID.  Only the owner or system profiles are accessible."""
    profile = db.query(ImapIngestionProfile).filter(ImapIngestionProfile.id == profile_id).first()
    if not profile or (profile.owner_id is not None and profile.owner_id != owner_id):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Ingestion profile not found")
    return _to_response(profile)


@router.put("/{profile_id}", summary="Update an ingestion profile")
def update_profile(
    profile_id: int,
    request: Request,
    body: ImapProfileUpdate,
    db: DbSession,
    owner_id: CurrentOwner,
) -> dict[str, Any]:
    """Update an existing ingestion profile.  Built-in profiles cannot be modified."""
    profile = db.query(ImapIngestionProfile).filter(ImapIngestionProfile.id == profile_id).first()
    if not profile or (profile.owner_id is not None and profile.owner_id != owner_id):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Ingestion profile not found")
    if profile.is_builtin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Built-in profiles cannot be modified.",
        )

    if body.name is not None:
        profile.name = body.name
    if "description" in body.model_fields_set:
        profile.description = body.description
    if body.allowed_categories is not None:
        categories = _validate_categories(body.allowed_categories)
        profile.allowed_categories = json.dumps(categories)

    profile.updated_at = datetime.now(timezone.utc)

    try:
        db.commit()
        db.refresh(profile)
    except Exception:
        db.rollback()
        raise

    logger.info("User %s updated IMAP ingestion profile %d", owner_id, profile_id)
    return _to_response(profile)


@router.delete("/{profile_id}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete an ingestion profile")
def delete_profile(profile_id: int, request: Request, db: DbSession, owner_id: CurrentOwner) -> None:
    """Delete an ingestion profile.  Built-in profiles cannot be deleted."""
    profile = db.query(ImapIngestionProfile).filter(ImapIngestionProfile.id == profile_id).first()
    if not profile or (profile.owner_id is not None and profile.owner_id != owner_id):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Ingestion profile not found")
    if profile.is_builtin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Built-in profiles cannot be deleted.",
        )

    try:
        db.delete(profile)
        db.commit()
    except Exception:
        db.rollback()
        raise

    logger.info("User %s deleted IMAP ingestion profile %d", owner_id, profile_id)