152 lines
No EOL
5 KiB
Python
152 lines
No EOL
5 KiB
Python
from fastapi import APIRouter, HTTPException, Depends, Body, Query, Path, Request
|
|
from typing import List, Dict, Any, Optional
|
|
from fastapi import status
|
|
from datetime import datetime
|
|
import logging
|
|
|
|
# Import auth module
|
|
from modules.security.auth import limiter, getCurrentUser
|
|
|
|
# Import interfaces
|
|
import modules.interfaces.serviceManagementClass as serviceManagementClass
|
|
from modules.interfaces.serviceManagementModel import Prompt
|
|
from modules.interfaces.serviceAppModel import AttributeDefinition, 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 = serviceManagementClass.getInterface(currentUser)
|
|
prompts = managementInterface.getAllPrompts()
|
|
return [Prompt.from_dict(prompt) for prompt in 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 = serviceManagementClass.getInterface(currentUser)
|
|
|
|
# Convert Prompt to dict for interface
|
|
prompt_data = prompt.to_dict()
|
|
|
|
# Create prompt
|
|
newPrompt = managementInterface.createPrompt(prompt_data)
|
|
|
|
# Set current time for createdAt if it exists in the model
|
|
if "createdAt" in Prompt.getModelAttributeDefinitions() and hasattr(newPrompt, "createdAt"):
|
|
newPrompt["createdAt"] = datetime.now().isoformat()
|
|
|
|
return Prompt.from_dict(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 = serviceManagementClass.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.from_dict(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 = serviceManagementClass.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
|
|
update_data = promptData.to_dict()
|
|
|
|
# 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.from_dict(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 = serviceManagementClass.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"}
|
|
|
|
@router.get("/attributes", response_model=List[AttributeDefinition])
|
|
@limiter.limit("30/minute")
|
|
async def get_prompt_attributes(
|
|
request: Request,
|
|
currentUser: User = Depends(getCurrentUser)
|
|
) -> List[AttributeDefinition]:
|
|
"""
|
|
Retrieves the attribute definitions for prompts.
|
|
This can be used for dynamic form generation.
|
|
|
|
Returns:
|
|
- A list of attribute definitions that can be used to generate forms
|
|
"""
|
|
# Get attributes from the Prompt model class
|
|
return Prompt.getModelAttributeDefinitions() |