forked from open-webui/open-webui
119 lines
3 KiB
Python
119 lines
3 KiB
Python
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.web.models.documents import (
|
|
Documents,
|
|
DocumentForm,
|
|
DocumentUpdateForm,
|
|
DocumentModel,
|
|
)
|
|
|
|
from utils.utils import get_current_user
|
|
from constants import ERROR_MESSAGES
|
|
|
|
router = APIRouter()
|
|
|
|
############################
|
|
# GetDocuments
|
|
############################
|
|
|
|
|
|
@router.get("/", response_model=List[DocumentModel])
|
|
async def get_documents(user=Depends(get_current_user)):
|
|
return Documents.get_docs()
|
|
|
|
|
|
############################
|
|
# CreateNewDoc
|
|
############################
|
|
|
|
|
|
@router.post("/create", response_model=Optional[DocumentModel])
|
|
async def create_new_doc(form_data: DocumentForm, user=Depends(get_current_user)):
|
|
if user.role != "admin":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
doc = Documents.get_doc_by_name(form_data.name)
|
|
if doc == None:
|
|
doc = Documents.insert_new_doc(user.id, form_data)
|
|
|
|
if doc:
|
|
return doc
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=ERROR_MESSAGES.FILE_EXISTS,
|
|
)
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=ERROR_MESSAGES.NAME_TAG_TAKEN,
|
|
)
|
|
|
|
|
|
############################
|
|
# GetDocByName
|
|
############################
|
|
|
|
|
|
@router.get("/name/{name}", response_model=Optional[DocumentModel])
|
|
async def get_doc_by_name(name: str, user=Depends(get_current_user)):
|
|
doc = Documents.get_doc_by_name(name)
|
|
|
|
if doc:
|
|
return doc
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.NOT_FOUND,
|
|
)
|
|
|
|
|
|
############################
|
|
# UpdateDocByName
|
|
############################
|
|
|
|
|
|
@router.post("/name/{name}/update", response_model=Optional[DocumentModel])
|
|
async def update_doc_by_name(
|
|
name: str, form_data: DocumentUpdateForm, user=Depends(get_current_user)
|
|
):
|
|
if user.role != "admin":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
doc = Documents.update_doc_by_name(name, form_data)
|
|
if doc:
|
|
return doc
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=ERROR_MESSAGES.NAME_TAG_TAKEN,
|
|
)
|
|
|
|
|
|
############################
|
|
# DeleteDocByName
|
|
############################
|
|
|
|
|
|
@router.delete("/name/{name}/delete", response_model=bool)
|
|
async def delete_doc_by_name(name: str, user=Depends(get_current_user)):
|
|
if user.role != "admin":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
result = Documents.delete_doc_by_name(name)
|
|
return result
|