gateway/modules/routes/routePrompts.py

132 lines
No EOL
4 KiB
Python

from fastapi import APIRouter, HTTPException, Depends, Body, Query, Path
from typing import List, Dict, Any, Optional
from fastapi import status
from datetime import datetime
# Import auth module
from modules.security.auth import getCurrentActiveUser
# Import interface
from modules.interfaces.lucydomInterface import getInterface
from modules.interfaces.lucydomModel import Prompt, getModelAttributes
# Model attributes for Prompt
promptAttributes = getModelAttributes(Prompt)
# Create router for prompt endpoints
router = APIRouter(
prefix="/api/prompts",
tags=["Prompts"],
responses={404: {"description": "Not found"}}
)
@router.get("", response_model=List[Dict[str, Any]])
async def get_prompts(
currentUser: Dict[str, Any] = Depends(getCurrentActiveUser)
):
"""Get all prompts"""
myInterface = getInterface(currentUser)
return myInterface.getAllPrompts()
@router.post("", response_model=Dict[str, Any])
async def create_prompt(
prompt: Dict[str, Any] = Body(...),
currentUser: Dict[str, Any] = Depends(getCurrentActiveUser)
):
"""Create a new prompt"""
myInterface = getInterface(currentUser)
# Required fields with default values
content = prompt.get("content", "")
name = prompt.get("name", "New Prompt")
# Create prompt
newPrompt = myInterface.createPrompt(
content=content,
name=name
)
# Set current time for createdAt if it exists in the model
if "createdAt" in promptAttributes and hasattr(newPrompt, "createdAt"):
newPrompt["createdAt"] = datetime.now().isoformat()
return newPrompt
@router.get("/{promptId}", response_model=Dict[str, Any])
async def get_prompt(
promptId: str = Path(..., description="ID of the prompt"),
currentUser: Dict[str, Any] = Depends(getCurrentActiveUser)
):
"""Get a specific prompt"""
myInterface = getInterface(currentUser)
# Get prompt
prompt = myInterface.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=Dict[str, Any])
async def update_prompt(
promptId: str = Path(..., description="ID of the prompt to update"),
promptData: Dict[str, Any] = Body(...),
currentUser: Dict[str, Any] = Depends(getCurrentActiveUser)
):
"""Update an existing prompt"""
myInterface = getInterface(currentUser)
# Check if the prompt exists
existingPrompt = myInterface.getPrompt(promptId)
if not existingPrompt:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Prompt with ID {promptId} not found"
)
# Standard fields for update
content = promptData.get("content")
name = promptData.get("name")
# Update prompt
updatedPrompt = myInterface.updatePrompt(
promptId=promptId,
content=content,
name=name
)
if not updatedPrompt:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Error updating the prompt"
)
return updatedPrompt
@router.delete("/{promptId}", response_model=Dict[str, Any])
async def delete_prompt(
promptId: str = Path(..., description="ID of the prompt to delete"),
currentUser: Dict[str, Any] = Depends(getCurrentActiveUser)
):
"""Delete a prompt"""
myInterface = getInterface(currentUser)
# Check if the prompt exists
existingPrompt = myInterface.getPrompt(promptId)
if not existingPrompt:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Prompt with ID {promptId} not found"
)
success = myInterface.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"}