105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
"""Admin API for Odysseus manifest addons."""
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from pydantic import BaseModel, Field
|
|
|
|
from src.addons import (
|
|
addon_template,
|
|
delete_addon,
|
|
get_addon,
|
|
get_enabled_addon_contributions,
|
|
list_addons,
|
|
save_addon_manifest,
|
|
set_addon_enabled,
|
|
)
|
|
|
|
|
|
class AddonManifestRequest(BaseModel):
|
|
manifest: dict = Field(default_factory=dict)
|
|
|
|
|
|
class AddonEnabledRequest(BaseModel):
|
|
enabled: bool
|
|
|
|
|
|
def setup_addon_routes(auth_manager) -> APIRouter:
|
|
router = APIRouter(prefix="/api/addons", tags=["addons"])
|
|
|
|
def _require_admin(request: Request) -> str:
|
|
token = request.cookies.get("odysseus_session")
|
|
user = auth_manager.get_username_for_token(token)
|
|
if not user:
|
|
raise HTTPException(401, "Not authenticated")
|
|
if not auth_manager.is_admin(user):
|
|
raise HTTPException(403, "Admin only")
|
|
return user
|
|
|
|
@router.get("")
|
|
async def list_all(request: Request):
|
|
_require_admin(request)
|
|
return {
|
|
"addons": list_addons(),
|
|
"enabled_contributions": get_enabled_addon_contributions(),
|
|
}
|
|
|
|
@router.get("/template")
|
|
async def get_template(request: Request, addon_id: str = "my-addon"):
|
|
_require_admin(request)
|
|
try:
|
|
return {"manifest": addon_template(addon_id)}
|
|
except ValueError as exc:
|
|
raise HTTPException(400, str(exc)) from exc
|
|
|
|
@router.post("")
|
|
async def create_or_update(body: AddonManifestRequest, request: Request):
|
|
_require_admin(request)
|
|
try:
|
|
addon = save_addon_manifest(body.manifest)
|
|
except ValueError as exc:
|
|
raise HTTPException(400, str(exc)) from exc
|
|
return {"ok": True, "addon": addon}
|
|
|
|
@router.get("/{addon_id}")
|
|
async def get_one(addon_id: str, request: Request):
|
|
_require_admin(request)
|
|
try:
|
|
addon = get_addon(addon_id)
|
|
except ValueError as exc:
|
|
raise HTTPException(400, str(exc)) from exc
|
|
if not addon:
|
|
raise HTTPException(404, "Addon not found")
|
|
return {"addon": addon}
|
|
|
|
@router.patch("/{addon_id}")
|
|
async def update_one(addon_id: str, body: AddonManifestRequest, request: Request):
|
|
_require_admin(request)
|
|
manifest = dict(body.manifest or {})
|
|
manifest["id"] = addon_id
|
|
try:
|
|
addon = save_addon_manifest(manifest)
|
|
except ValueError as exc:
|
|
raise HTTPException(400, str(exc)) from exc
|
|
return {"ok": True, "addon": addon}
|
|
|
|
@router.post("/{addon_id}/enabled")
|
|
async def set_enabled(addon_id: str, body: AddonEnabledRequest, request: Request):
|
|
_require_admin(request)
|
|
try:
|
|
addon = set_addon_enabled(addon_id, body.enabled)
|
|
except KeyError as exc:
|
|
raise HTTPException(404, "Addon not found") from exc
|
|
except ValueError as exc:
|
|
raise HTTPException(400, str(exc)) from exc
|
|
return {"ok": True, "addon": addon}
|
|
|
|
@router.delete("/{addon_id}")
|
|
async def delete_one(addon_id: str, request: Request):
|
|
_require_admin(request)
|
|
try:
|
|
deleted = delete_addon(addon_id)
|
|
except ValueError as exc:
|
|
raise HTTPException(400, str(exc)) from exc
|
|
return {"ok": deleted}
|
|
|
|
return router
|