Lars
d6838abda7
Unmapped ohne Kreuzjoin, Listen-GET kurz, Rebuild im Hintergrund. Tandoor-Zugang (URL/Token) in den Einstellungen mit ausliefern. Co-authored-by: Cursor <cursoragent@cursor.com>
79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
"""Per-user Tandoor connection. No recipe import yet."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
from starlette.requests import Request
|
|
from slowapi import Limiter
|
|
from slowapi.util import get_remote_address
|
|
|
|
from auth import require_auth
|
|
from data_layer.tandoor_client import TandoorError, normalize_base_url, probe_connection
|
|
from data_layer.tandoor_settings import (
|
|
get_public_settings,
|
|
get_settings,
|
|
record_probe,
|
|
upsert_settings,
|
|
)
|
|
from db import get_cursor, get_db
|
|
|
|
router = APIRouter(prefix="/api/tandoor", tags=["tandoor"])
|
|
limiter = Limiter(key_func=get_remote_address)
|
|
|
|
|
|
class TandoorSettingsIn(BaseModel):
|
|
base_url: str = ""
|
|
token: Optional[str] = None
|
|
clear_token: bool = False
|
|
|
|
|
|
class TandoorProbeIn(BaseModel):
|
|
base_url: Optional[str] = None
|
|
token: Optional[str] = Field(default=None, description="Nur zum Testen; leer = gespeichertes Token")
|
|
|
|
|
|
@router.get("/settings")
|
|
def read_settings(session: dict = Depends(require_auth)):
|
|
pid = session["profile_id"]
|
|
with get_db() as conn:
|
|
return get_public_settings(get_cursor(conn), pid)
|
|
|
|
|
|
@router.put("/settings")
|
|
def write_settings(body: TandoorSettingsIn, session: dict = Depends(require_auth)):
|
|
pid = session["profile_id"]
|
|
try:
|
|
if body.clear_token:
|
|
url = ""
|
|
else:
|
|
url = normalize_base_url(body.base_url)
|
|
with get_db() as conn:
|
|
return upsert_settings(
|
|
get_cursor(conn), pid, url, body.token, clear_token=body.clear_token
|
|
)
|
|
except (TandoorError, ValueError) as e:
|
|
raise HTTPException(400, str(e)) from e
|
|
|
|
|
|
@router.post("/test")
|
|
@limiter.limit("10/minute")
|
|
def test_connection(request: Request, body: TandoorProbeIn | None = None, session: dict = Depends(require_auth)):
|
|
pid = session["profile_id"]
|
|
body = body or TandoorProbeIn()
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
stored = get_settings(cur, pid)
|
|
url = (body.base_url or "").strip() or ((stored or {}).get("base_url") or "")
|
|
token = (body.token or "").strip() or ((stored or {}).get("api_token") or "")
|
|
try:
|
|
result = probe_connection(url, token)
|
|
if stored:
|
|
record_probe(cur, pid, True)
|
|
return result
|
|
except TandoorError as e:
|
|
if stored:
|
|
record_probe(cur, pid, False, str(e))
|
|
raise HTTPException(400, str(e)) from e
|