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.interfaceDbComponentObjects as interfaceDbComponentObjects
from modules.datamodels.datamodelUtils import Prompt
from modules.datamodels.datamodelUam import User
from modules.datamodels.datamodelPagination import PaginationParams, PaginatedResponse, PaginationMetadata, normalize_pagination_dict

# 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")
async 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 = interfaceDbComponentObjects.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.post("", response_model=Prompt)
@limiter.limit("10/minute")
async def create_prompt(
    request: Request,
    prompt: Prompt,
    currentUser: User = Depends(getCurrentUser)
) -> Prompt:
    """Create a new prompt"""
    managementInterface = interfaceDbComponentObjects.getInterface(currentUser)
    
    # Create prompt
    newPrompt = managementInterface.createPrompt(prompt)
    
    return Prompt(**newPrompt)

@router.get("/{promptId}", response_model=Prompt)
@limiter.limit("30/minute")
async def get_prompt(
    request: Request,
    promptId: str = Path(..., description="ID of the prompt"),
    currentUser: User = Depends(getCurrentUser)
) -> Prompt:
    """Get a specific prompt"""
    managementInterface = interfaceDbComponentObjects.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")
async def update_prompt(
    request: Request,
    promptId: str = Path(..., description="ID of the prompt to update"),
    promptData: Prompt = Body(...),
    currentUser: User = Depends(getCurrentUser)
) -> Prompt:
    """Update an existing prompt"""
    managementInterface = interfaceDbComponentObjects.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"
        )
    
    # Convert Prompt to dict for interface, excluding the id field
    if hasattr(promptData, "model_dump"):
        update_data = promptData.model_dump(exclude={"id"})
    else:
        update_data = promptData.model_dump(exclude={"id"})
    
    # Update prompt
    updatedPrompt = managementInterface.updatePrompt(promptId, update_data)
    
    if not updatedPrompt:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Error updating the prompt"
        )
    
    return Prompt(**updatedPrompt)

@router.delete("/{promptId}", response_model=Dict[str, Any])
@limiter.limit("10/minute")
async 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 = interfaceDbComponentObjects.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"
        )
    
    success = managementInterface.deletePrompt(promptId)
    if not success:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Error deleting the prompt"
        )
    
    return {"message": f"Prompt with ID {promptId} successfully deleted"}