Spaces:
Runtime error
Runtime error
File size: 3,742 Bytes
9e93b10 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 | from __future__ import annotations
import logging
import black
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from rexpro_ai.config import DATA_DIR, ENABLE_ADMIN_EXPORT
from rexpro_ai.constants import ERROR_MESSAGES
from rexpro_ai.models.chats import ChatTitleMessagesForm
from rexpro_ai.utils.auth import get_admin_user, get_verified_user
from rexpro_ai.utils.code_interpreter import execute_code_jupyter
from rexpro_ai.utils.misc import get_gravatar_url
from rexpro_ai.utils.pdf_generator import PDFGenerator
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 request.app.state.config.ENABLE_CODE_EXECUTION:
raise HTTPException(
status_code=403,
detail=ERROR_MESSAGES.FEATURE_DISABLED('Code execution'),
)
if request.app.state.config.CODE_EXECUTION_ENGINE == 'jupyter':
output = await execute_code_jupyter(
request.app.state.config.CODE_EXECUTION_JUPYTER_URL,
form_data.code,
(
request.app.state.config.CODE_EXECUTION_JUPYTER_AUTH_TOKEN
if request.app.state.config.CODE_EXECUTION_JUPYTER_AUTH == 'token'
else None
),
(
request.app.state.config.CODE_EXECUTION_JUPYTER_AUTH_PASSWORD
if request.app.state.config.CODE_EXECUTION_JUPYTER_AUTH == 'password'
else None
),
request.app.state.config.CODE_EXECUTION_JUPYTER_TIMEOUT,
)
return output
else:
raise HTTPException(
status_code=400,
detail=ERROR_MESSAGES.DEFAULT('Code execution engine not supported'),
)
class ChatForm(BaseModel):
title: str
messages: list[dict]
@router.post('/pdf')
async def download_chat_as_pdf(form_data: ChatTitleMessagesForm, user=Depends(get_verified_user)):
try:
pdf_bytes = PDFGenerator(form_data).generate_chat_pdf()
return Response(
content=pdf_bytes,
media_type='application/pdf',
headers={'Content-Disposition': 'attachment;filename=chat.pdf'},
)
except Exception as e:
log.exception(f'Error generating PDF: {e}')
raise HTTPException(status_code=400, detail=str(e))
@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 rexpro_ai.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='rexpro.db',
)
|