122 lines
3.9 KiB
Python
122 lines
3.9 KiB
Python
"""
|
|
FastAPI adapter for the Gitea-backed suggestions module.
|
|
|
|
Use this INSTEAD of routes_flask.py if the app is FastAPI. Wire it in with:
|
|
|
|
from suggestions.routes_fastapi import router as suggestions_router
|
|
app.include_router(suggestions_router)
|
|
|
|
AUTH: replace `current_user()` with the app's real dependency. It must return
|
|
something exposing `.name` and `.email`, or None.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, File, Request, UploadFile
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from .gitea import (
|
|
ALLOWED_IMAGE_TYPES,
|
|
MAX_UPLOAD_BYTES,
|
|
GiteaClient,
|
|
GiteaError,
|
|
footer,
|
|
gitea_config,
|
|
save_upload,
|
|
sniff_image,
|
|
)
|
|
|
|
router = APIRouter()
|
|
UPLOAD_DIR = Path("/data/uploads/suggestions")
|
|
|
|
|
|
def current_user():
|
|
"""ADAPT THIS to the app's real auth. Return obj with .name/.email or None."""
|
|
return type("U", (), {"name": "Household", "email": "household@local"})()
|
|
|
|
|
|
def _client_or_error():
|
|
cfg = gitea_config()
|
|
if not cfg:
|
|
return None, JSONResponse({"error": "not_configured"}, status_code=503)
|
|
return GiteaClient(cfg), None
|
|
|
|
|
|
@router.get("/api/suggestions")
|
|
def list_suggestions(state: str = "open"):
|
|
if current_user() is None:
|
|
return JSONResponse({"error": "unauthorized"}, status_code=401)
|
|
client, err = _client_or_error()
|
|
if err:
|
|
return err
|
|
try:
|
|
return {"issues": client.list_issues(state)}
|
|
except GiteaError as e:
|
|
return JSONResponse({"error": str(e)}, status_code=502)
|
|
|
|
|
|
@router.post("/api/suggestions")
|
|
async def create_suggestion(request: Request):
|
|
user = current_user()
|
|
if user is None:
|
|
return JSONResponse({"error": "unauthorized"}, status_code=401)
|
|
client, err = _client_or_error()
|
|
if err:
|
|
return err
|
|
data = await request.json()
|
|
title = (data.get("title") or "").strip()
|
|
if len(title) < 3:
|
|
return JSONResponse({"error": "title_required"}, status_code=400)
|
|
body = (data.get("body") or "").strip() + footer(user.name, user.email)
|
|
try:
|
|
return JSONResponse({"issue": client.create_issue(title, body)}, status_code=201)
|
|
except GiteaError as e:
|
|
return JSONResponse({"error": str(e)}, status_code=502)
|
|
|
|
|
|
@router.get("/api/suggestions/{number}/comments")
|
|
def list_comments(number: int):
|
|
if current_user() is None:
|
|
return JSONResponse({"error": "unauthorized"}, status_code=401)
|
|
client, err = _client_or_error()
|
|
if err:
|
|
return err
|
|
try:
|
|
return {"comments": client.list_comments(number)}
|
|
except GiteaError as e:
|
|
return JSONResponse({"error": str(e)}, status_code=502)
|
|
|
|
|
|
@router.post("/api/suggestions/{number}/comments")
|
|
async def add_comment(number: int, request: Request):
|
|
user = current_user()
|
|
if user is None:
|
|
return JSONResponse({"error": "unauthorized"}, status_code=401)
|
|
client, err = _client_or_error()
|
|
if err:
|
|
return err
|
|
data = await request.json()
|
|
text = (data.get("body") or "").strip()
|
|
if not text:
|
|
return JSONResponse({"error": "body_required"}, status_code=400)
|
|
text += f"\n\n_— {user.name}_"
|
|
try:
|
|
return JSONResponse({"comment": client.create_comment(number, text)}, status_code=201)
|
|
except GiteaError as e:
|
|
return JSONResponse({"error": str(e)}, status_code=502)
|
|
|
|
|
|
@router.post("/api/suggestions/upload")
|
|
async def upload(request: Request, file: UploadFile = File(...)):
|
|
if current_user() is None:
|
|
return JSONResponse({"error": "unauthorized"}, status_code=401)
|
|
data = await file.read()
|
|
if len(data) > MAX_UPLOAD_BYTES:
|
|
return JSONResponse({"error": "too_large"}, status_code=413)
|
|
if sniff_image(data) not in ALLOWED_IMAGE_TYPES:
|
|
return JSONResponse({"error": "unsupported_type"}, status_code=415)
|
|
origin = str(request.base_url)
|
|
url = save_upload(file.filename or "image", data, UPLOAD_DIR, origin)
|
|
return JSONResponse({"url": url}, status_code=201)
|