|
from fastapi import Depends, FastAPI, HTTPException, status |
|
from datetime import datetime, timedelta |
|
from typing import List, Union, Optional |
|
|
|
from fastapi import APIRouter |
|
from pydantic import BaseModel |
|
import json |
|
|
|
from apps.webui.models.prompts import Prompts, PromptForm, PromptModel |
|
|
|
from utils.utils import get_current_user, get_admin_user |
|
from constants import ERROR_MESSAGES |
|
|
|
router = APIRouter() |
|
|
|
|
|
|
|
|
|
|
|
|
|
@router.get("/", response_model=List[PromptModel]) |
|
async def get_prompts(user=Depends(get_current_user)): |
|
return Prompts.get_prompts() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@router.post("/create", response_model=Optional[PromptModel]) |
|
async def create_new_prompt(form_data: PromptForm, user=Depends(get_admin_user)): |
|
prompt = Prompts.get_prompt_by_command(form_data.command) |
|
if prompt == None: |
|
prompt = Prompts.insert_new_prompt(user.id, form_data) |
|
|
|
if prompt: |
|
return prompt |
|
raise HTTPException( |
|
status_code=status.HTTP_400_BAD_REQUEST, |
|
detail=ERROR_MESSAGES.DEFAULT(), |
|
) |
|
raise HTTPException( |
|
status_code=status.HTTP_400_BAD_REQUEST, |
|
detail=ERROR_MESSAGES.COMMAND_TAKEN, |
|
) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@router.get("/command/{command}", response_model=Optional[PromptModel]) |
|
async def get_prompt_by_command(command: str, user=Depends(get_current_user)): |
|
prompt = Prompts.get_prompt_by_command(f"/{command}") |
|
|
|
if prompt: |
|
return prompt |
|
else: |
|
raise HTTPException( |
|
status_code=status.HTTP_401_UNAUTHORIZED, |
|
detail=ERROR_MESSAGES.NOT_FOUND, |
|
) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@router.post("/command/{command}/update", response_model=Optional[PromptModel]) |
|
async def update_prompt_by_command( |
|
command: str, form_data: PromptForm, user=Depends(get_admin_user) |
|
): |
|
prompt = Prompts.update_prompt_by_command(f"/{command}", form_data) |
|
if prompt: |
|
return prompt |
|
else: |
|
raise HTTPException( |
|
status_code=status.HTTP_401_UNAUTHORIZED, |
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED, |
|
) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@router.delete("/command/{command}/delete", response_model=bool) |
|
async def delete_prompt_by_command(command: str, user=Depends(get_admin_user)): |
|
result = Prompts.delete_prompt_by_command(f"/{command}") |
|
return result |
|
|