160 lines
No EOL
4.8 KiB
Python
160 lines
No EOL
4.8 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
|
|
from dataclasses import dataclass
|
|
|
|
# Import auth module
|
|
from modules.security.auth import getCurrentActiveUser, getUserContext
|
|
|
|
# Import interfaces
|
|
from modules.interfaces.lucydomInterface import getLucydomInterface
|
|
from modules.interfaces.lucydomModel import Prompt
|
|
|
|
# Get all attributes of the model
|
|
def getModelAttributes(modelClass):
|
|
return [attr for attr in dir(modelClass)
|
|
if not callable(getattr(modelClass, attr))
|
|
and not attr.startswith('_')
|
|
and attr not in ('metadata', 'query', 'query_class', 'label', 'field_labels')]
|
|
|
|
# Model attributes for Prompt
|
|
promptAttributes = getModelAttributes(Prompt)
|
|
|
|
@dataclass
|
|
class AppContext:
|
|
"""Context object for all required connections and user information"""
|
|
mandateId: int
|
|
userId: int
|
|
interfaceData: Any # LucyDOM Interface
|
|
|
|
async def getContext(currentUser: Dict[str, Any]) -> AppContext:
|
|
"""Creates a central context object with all required connections"""
|
|
mandateId, userId = await getUserContext(currentUser)
|
|
interfaceData = getLucydomInterface(mandateId, userId)
|
|
|
|
return AppContext(
|
|
mandateId=mandateId,
|
|
userId=userId,
|
|
interfaceData=interfaceData
|
|
)
|
|
|
|
# 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 getPrompts(
|
|
currentUser: Dict[str, Any] = Depends(getCurrentActiveUser)
|
|
):
|
|
"""Get all prompts"""
|
|
context = await getContext(currentUser)
|
|
|
|
# Retrieve prompts
|
|
return context.interfaceData.getAllPrompts()
|
|
|
|
@router.post("", response_model=Dict[str, Any])
|
|
async def createPrompt(
|
|
prompt: Dict[str, Any] = Body(...),
|
|
currentUser: Dict[str, Any] = Depends(getCurrentActiveUser)
|
|
):
|
|
"""Create a new prompt"""
|
|
context = await getContext(currentUser)
|
|
|
|
# Required fields with default values
|
|
content = prompt.get("content", "")
|
|
name = prompt.get("name", "New Prompt")
|
|
|
|
# Create prompt
|
|
newPrompt = context.interfaceData.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 getPrompt(
|
|
promptId: int,
|
|
currentUser: Dict[str, Any] = Depends(getCurrentActiveUser)
|
|
):
|
|
"""Get a specific prompt"""
|
|
context = await getContext(currentUser)
|
|
|
|
# Get prompt
|
|
prompt = context.interfaceData.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 updatePrompt(
|
|
promptId: int,
|
|
promptData: Dict[str, Any] = Body(...),
|
|
currentUser: Dict[str, Any] = Depends(getCurrentActiveUser)
|
|
):
|
|
"""Update an existing prompt"""
|
|
context = await getContext(currentUser)
|
|
|
|
# Check if the prompt exists
|
|
existingPrompt = context.interfaceData.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 = context.interfaceData.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 deletePrompt(
|
|
promptId: int,
|
|
currentUser: Dict[str, Any] = Depends(getCurrentActiveUser)
|
|
):
|
|
"""Delete a prompt"""
|
|
context = await getContext(currentUser)
|
|
|
|
# Check if the prompt exists
|
|
existingPrompt = context.interfaceData.getPrompt(promptId)
|
|
if not existingPrompt:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Prompt with ID {promptId} not found"
|
|
)
|
|
|
|
success = context.interfaceData.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"} |