mirror of
https://github.com/open-webui/open-webui.git
synced 2026-09-16 23:43:03 +00:00
90 lines
3 KiB
Python
90 lines
3 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
import black
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from open_webui.config import DATA_DIR, ENABLE_ADMIN_EXPORT
|
|
from open_webui.constants import ERROR_MESSAGES
|
|
from open_webui.models.config import Config
|
|
from open_webui.utils.auth import get_admin_user, get_verified_user
|
|
from open_webui.utils.code_interpreter import execute_code_jupyter
|
|
from open_webui.utils.misc import get_gravatar_url
|
|
from pydantic import BaseModel
|
|
from starlette.responses import FileResponse
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get('/gravatar')
|
|
async def get_gravatar(email: str, user=Depends(get_verified_user)):
|
|
return get_gravatar_url(email)
|
|
|
|
|
|
class CodeForm(BaseModel):
|
|
code: str
|
|
|
|
|
|
@router.post('/code/format')
|
|
async def format_code(form_data: CodeForm, user=Depends(get_admin_user)):
|
|
try:
|
|
formatted_code = black.format_str(form_data.code, mode=black.Mode())
|
|
return {'code': formatted_code}
|
|
except black.NothingChanged:
|
|
return {'code': form_data.code}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@router.post('/code/execute')
|
|
async def execute_code(request: Request, form_data: CodeForm, user=Depends(get_verified_user)):
|
|
if not await Config.get('code_execution.enable'):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=ERROR_MESSAGES.FEATURE_DISABLED('Code execution'),
|
|
)
|
|
|
|
if await Config.get('code_execution.engine') == 'jupyter':
|
|
output = await execute_code_jupyter(
|
|
await Config.get('code_execution.jupyter.url'),
|
|
form_data.code,
|
|
(
|
|
await Config.get('code_execution.jupyter.auth_token')
|
|
if await Config.get('code_execution.jupyter.auth') == 'token'
|
|
else None
|
|
),
|
|
(
|
|
await Config.get('code_execution.jupyter.auth_password')
|
|
if await Config.get('code_execution.jupyter.auth') == 'password'
|
|
else None
|
|
),
|
|
await Config.get('code_execution.jupyter.timeout'),
|
|
)
|
|
|
|
return output
|
|
else:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=ERROR_MESSAGES.DEFAULT('Code execution engine not supported'),
|
|
)
|
|
|
|
|
|
@router.get('/db/download')
|
|
async def download_db(user=Depends(get_admin_user)):
|
|
"""Download the raw SQLite database file (admin-only, SQLite deployments only)."""
|
|
if not ENABLE_ADMIN_EXPORT:
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.ACCESS_PROHIBITED)
|
|
|
|
# Lazy import avoids circular dependency at module load time
|
|
from open_webui.internal.db import engine
|
|
|
|
if engine.name != 'sqlite':
|
|
raise HTTPException(status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DB_NOT_SQLITE)
|
|
|
|
return FileResponse(
|
|
str(engine.url.database),
|
|
media_type='application/octet-stream',
|
|
filename='webui.db',
|
|
)
|