mirror of
https://github.com/open-webui/open-webui
synced 2024-11-07 09:09:53 +00:00
146 lines
3.4 KiB
Python
146 lines
3.4 KiB
Python
from fastapi import (
|
|
Depends,
|
|
FastAPI,
|
|
HTTPException,
|
|
status,
|
|
Request,
|
|
UploadFile,
|
|
File,
|
|
Form,
|
|
)
|
|
|
|
|
|
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.files import Files, FileForm, FileModel, FileResponse
|
|
from utils.utils import get_verified_user, get_admin_user
|
|
from constants import ERROR_MESSAGES
|
|
|
|
from importlib import util
|
|
import os
|
|
import uuid
|
|
|
|
from config import SRC_LOG_LEVELS, UPLOAD_DIR
|
|
|
|
|
|
import logging
|
|
|
|
log = logging.getLogger(__name__)
|
|
log.setLevel(SRC_LOG_LEVELS["MODELS"])
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
############################
|
|
# Upload File
|
|
############################
|
|
|
|
|
|
@router.post("/")
|
|
def upload_file(
|
|
file: UploadFile = File(...),
|
|
user=Depends(get_verified_user),
|
|
):
|
|
log.info(f"file.content_type: {file.content_type}")
|
|
try:
|
|
unsanitized_filename = file.filename
|
|
filename = os.path.basename(unsanitized_filename)
|
|
|
|
# replace filename with uuid
|
|
id = str(uuid.uuid4())
|
|
file_path = f"{UPLOAD_DIR}/{filename}"
|
|
|
|
contents = file.file.read()
|
|
with open(file_path, "wb") as f:
|
|
f.write(contents)
|
|
f.close()
|
|
|
|
file = Files.insert_new_file(
|
|
user.id,
|
|
FileForm(
|
|
**{
|
|
"id": id,
|
|
"filename": filename,
|
|
"meta": {
|
|
"content_type": file.content_type,
|
|
"size": len(contents),
|
|
"path": file_path,
|
|
},
|
|
}
|
|
),
|
|
)
|
|
|
|
if file:
|
|
return file
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=ERROR_MESSAGES.DEFAULT("Error uploading file"),
|
|
)
|
|
|
|
except Exception as e:
|
|
log.exception(e)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=ERROR_MESSAGES.DEFAULT(e),
|
|
)
|
|
|
|
|
|
############################
|
|
# List Files
|
|
############################
|
|
|
|
|
|
@router.get("/", response_model=List[FileModel])
|
|
async def list_files(user=Depends(get_verified_user)):
|
|
files = Files.get_files()
|
|
return files
|
|
|
|
|
|
############################
|
|
# Get File By Id
|
|
############################
|
|
|
|
|
|
@router.get("/{id}", response_model=Optional[FileModel])
|
|
async def get_file_by_id(id: str, user=Depends(get_verified_user)):
|
|
file = Files.get_file_by_id(id)
|
|
|
|
if file:
|
|
return file
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.NOT_FOUND,
|
|
)
|
|
|
|
|
|
############################
|
|
# Delete File By Id
|
|
############################
|
|
|
|
|
|
@router.delete("/{id}")
|
|
async def delete_file_by_id(id: str, user=Depends(get_verified_user)):
|
|
file = Files.get_file_by_id(id)
|
|
|
|
if file:
|
|
result = Files.delete_file_by_id(id)
|
|
if result:
|
|
return {"message": "File deleted successfully"}
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=ERROR_MESSAGES.DEFAULT("Error deleting file"),
|
|
)
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.NOT_FOUND,
|
|
)
|