from fastapi import APIRouter, HTTPException, Depends, Body, Query, Path, Request, Response from typing import List, Dict, Any, Optional from fastapi import status from datetime import datetime import logging import inspect import importlib import os from pydantic import BaseModel # Import auth module from modules.security.auth import limiter, getCurrentUser # Import interfaces import modules.interfaces.interfaceDbComponentObjects as interfaceDbComponentObjects from modules.datamodels.datamodelUtils import Prompt from modules.shared.attributeUtils import getModelAttributeDefinitions, AttributeResponse, AttributeDefinition from modules.datamodels.datamodelUam import User # 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=List[Prompt]) @limiter.limit("30/minute") async def get_prompts( request: Request, currentUser: User = Depends(getCurrentUser) ) -> List[Prompt]: """Get all prompts""" managementInterface = interfaceDbComponentObjects.getInterface(currentUser) prompts = managementInterface.getAllPrompts() return prompts @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"}