gateway/modules/routes/routeDataPrompts.py

# Copyright (c) 2025 Patrick Motsch
# All rights reserved.
from fastapi import APIRouter, HTTPException, Depends, Body, Path, Request, Query
from typing import List, Dict, Any, Optional
from fastapi import status
import logging
import json

# Import auth module
from modules.auth import limiter, getCurrentUser

# Import interfaces
import modules.interfaces.interfaceDbManagement as interfaceDbManagement
from modules.datamodels.datamodelUtils import Prompt
from modules.datamodels.datamodelUam import User
from modules.datamodels.datamodelPagination import PaginationParams, PaginatedResponse, PaginationMetadata, normalize_pagination_dict
from modules.shared.i18nRegistry import apiRouteContext
routeApiMsg = apiRouteContext("routeDataPrompts")

# Configure logger
logger = logging.getLogger(__name__)

# Create router for prompt endpoints
router = APIRouter(
    prefix="/api/prompts",
    tags=["Manage Prompts"],
    responses={404: {"description": "Not found"}}
)

@router.get("", response_model=PaginatedResponse[Prompt])
@limiter.limit("30/minute")
def get_prompts(
    request: Request,
    pagination: Optional[str] = Query(None, description="JSON-encoded PaginationParams object"),
    currentUser: User = Depends(getCurrentUser)
) -> PaginatedResponse[Prompt]:
    """
    Get prompts with optional pagination, sorting, and filtering.
    
    Query Parameters:
    - pagination: JSON-encoded PaginationParams object, or None for no pagination
    
    Examples:
    - GET /api/prompts (no pagination - returns all items)
    - GET /api/prompts?pagination={"page":1,"pageSize":10,"sort":[]}
    - GET /api/prompts?pagination={"page":2,"pageSize":20,"sort":[{"field":"name","direction":"asc"}]}
    """
    # Parse pagination parameter
    paginationParams = None
    if pagination:
        try:
            paginationDict = json.loads(pagination)
            if paginationDict:
                # Normalize pagination dict (handles top-level "search" field)
                paginationDict = normalize_pagination_dict(paginationDict)
                paginationParams = PaginationParams(**paginationDict)
        except (json.JSONDecodeError, ValueError) as e:
            raise HTTPException(
                status_code=400,
                detail=f"Invalid pagination parameter: {str(e)}"
            )
    
    managementInterface = interfaceDbManagement.getInterface(currentUser)
    result = managementInterface.getAllPrompts(pagination=paginationParams)
    
    # If pagination was requested, result is PaginatedResult
    # If no pagination, result is List[Prompt]
    if paginationParams:
        return PaginatedResponse(
            items=result.items,
            pagination=PaginationMetadata(
                currentPage=paginationParams.page,
                pageSize=paginationParams.pageSize,
                totalItems=result.totalItems,
                totalPages=result.totalPages,
                sort=paginationParams.sort,
                filters=paginationParams.filters
            )
        )
    else:
        return PaginatedResponse(
            items=result,
            pagination=None
        )

@router.get("/filter-values")
@limiter.limit("60/minute")
def get_prompt_filter_values(
    request: Request,
    column: str = Query(..., description="Column key"),
    pagination: Optional[str] = Query(None, description="JSON-encoded current filters"),
    currentUser: User = Depends(getCurrentUser)
) -> list:
    """Return distinct filter values for a column in prompts.
    
    NOTE: Cannot use db.getDistinctColumnValues() because visibility rules
    (own + system for regular users) require pre-filtering the recordset.
    """
    try:
        from modules.routes.routeDataUsers import _handleFilterValuesRequest
        managementInterface = interfaceDbManagement.getInterface(currentUser)
        result = managementInterface.getAllPrompts(pagination=None)
        items = [r.model_dump() if hasattr(r, 'model_dump') else r for r in result]
        return _handleFilterValuesRequest(items, column, pagination)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@router.post("", response_model=Prompt)
@limiter.limit("10/minute")
def create_prompt(
    request: Request,
    prompt: Prompt,
    currentUser: User = Depends(getCurrentUser)
) -> Prompt:
    """Create a new prompt"""
    managementInterface = interfaceDbManagement.getInterface(currentUser)
    
    # Create prompt
    newPrompt = managementInterface.createPrompt(prompt)
    
    return Prompt(**newPrompt)

@router.get("/{promptId}", response_model=Prompt)
@limiter.limit("30/minute")
def get_prompt(
    request: Request,
    promptId: str = Path(..., description="ID of the prompt"),
    currentUser: User = Depends(getCurrentUser)
) -> Prompt:
    """Get a specific prompt"""
    managementInterface = interfaceDbManagement.getInterface(currentUser)
    
    # Get prompt
    prompt = managementInterface.getPrompt(promptId)
    if not prompt:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Prompt with ID {promptId} not found"
        )
    
    return prompt

@router.put("/{promptId}", response_model=Prompt)
@limiter.limit("10/minute")
def update_prompt(
    request: Request,
    promptId: str = Path(..., description="ID of the prompt to update"),
    promptData: Dict[str, Any] = Body(...),
    currentUser: User = Depends(getCurrentUser)
) -> Prompt:
    """Update an existing prompt (supports partial updates for inline editing)"""
    managementInterface = interfaceDbManagement.getInterface(currentUser)
    
    # Check if the prompt exists
    existingPrompt = managementInterface.getPrompt(promptId)
    if not existingPrompt:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Prompt with ID {promptId} not found"
        )
    
    # Remove id from update data if present
    update_data = {k: v for k, v in promptData.items() if k != "id"}
    
    # Update prompt (ownership check happens in interface)
    try:
        updatedPrompt = managementInterface.updatePrompt(promptId, update_data)
    except PermissionError as e:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=str(e)
        )
    
    if not updatedPrompt:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=routeApiMsg("Error updating the prompt")
        )
    
    return Prompt(**updatedPrompt)

@router.delete("/{promptId}", response_model=Dict[str, Any])
@limiter.limit("10/minute")
def delete_prompt(
    request: Request,
    promptId: str = Path(..., description="ID of the prompt to delete"),
    currentUser: User = Depends(getCurrentUser)
) -> Dict[str, Any]:
    """Delete a prompt"""
    managementInterface = interfaceDbManagement.getInterface(currentUser)
    
    # Check if the prompt exists
    existingPrompt = managementInterface.getPrompt(promptId)
    if not existingPrompt:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Prompt with ID {promptId} not found"
        )
    
    try:
        success = managementInterface.deletePrompt(promptId)
    except PermissionError as e:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=str(e)
        )
    
    if not success:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=routeApiMsg("Error deleting the prompt")
        )
    
    return {"message": f"Prompt with ID {promptId} successfully deleted"}