Hitonabi
47f7a85510
Die Ampel-Nachruestung deckte 317/337 vorbestehende ruff-Verstoesse im ganzen Repo auf. Aufgeraeumt: - ruff.toml: intentionale Muster als Projekt-Politik ausgenommen (BLE001 blind-except, S110/S112 try-except-pass/continue, PLW1510 subprocess-best-effort, B008 FastAPI- Depends/File-Idiom, EXE001 Shebang, + wenige Stil-Regeln). __init__.py-Re-Exports geschuetzt (F401). - ruff --fix: 128 mechanische (Import-Sortierung, PEP585/604-Annotationen, tote Imports, ueberfluessige noqa) auto-behoben. - 12 echte Reste von Hand: PERF402/102, PLC3002 (Lambda->walrus), ISC004 (String-Concat geklammert), F841/RUF059 (ungenutzte Vars), PIE810 (startswith-Tuple), UP031 (f-string), UP035 (veraltete typing-Imports). Ergebnis: 'ruff check .' = 0, 'compileall' grün. Kein Verhaltenswechsel (nur Stil/Modernisierung). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
85 lines
2.4 KiB
Python
85 lines
2.4 KiB
Python
"""Wartungs-Endpoints: Update-Badge, OS-/Engine-Update, Reboot, Restart, Logs."""
|
|
|
|
from fastapi import APIRouter, Header, HTTPException
|
|
from pydantic import BaseModel
|
|
from services import maintenance
|
|
|
|
router = APIRouter(prefix="/api")
|
|
|
|
|
|
class SudoReq(BaseModel):
|
|
sudo_password: str | None = None
|
|
|
|
|
|
class RestartReq(BaseModel):
|
|
service: str
|
|
sudo_password: str | None = None
|
|
|
|
|
|
@router.get("/maintenance/updates")
|
|
def updates() -> dict:
|
|
return maintenance.updates()
|
|
|
|
|
|
@router.get("/maintenance/update-details")
|
|
def update_details(kind: str) -> dict:
|
|
if kind not in ("os", "engine", "swap", "hermes"):
|
|
raise HTTPException(400, "Unbekannte Update-Art.")
|
|
return maintenance.update_details(kind)
|
|
|
|
@router.post("/maintenance/check-updates")
|
|
def check_updates(body: SudoReq) -> dict:
|
|
res = maintenance.check_updates_job(body.sudo_password)
|
|
if isinstance(res, dict) and not res.get("ok", True):
|
|
return res
|
|
return res
|
|
|
|
|
|
@router.post("/maintenance/os-update")
|
|
def os_update(body: SudoReq) -> dict:
|
|
res = maintenance.os_update_job(body.sudo_password)
|
|
if isinstance(res, dict) and not res.get("ok", True):
|
|
return res
|
|
return res
|
|
|
|
|
|
@router.post("/maintenance/engine-update")
|
|
def engine_update(body: SudoReq) -> dict:
|
|
res = maintenance.engine_update_job(body.sudo_password)
|
|
if not res:
|
|
raise HTTPException(400, "Kein Engine-Update-Befehl gesetzt (MC_ENGINE_UPDATE_CMD).")
|
|
return res
|
|
|
|
|
|
@router.post("/maintenance/swap-update")
|
|
def swap_update(body: SudoReq) -> dict:
|
|
res = maintenance.swap_update_job(body.sudo_password)
|
|
if not res:
|
|
raise HTTPException(400, "Kein Router-Update-Befehl gesetzt (MC_SWAP_UPDATE_CMD).")
|
|
return res
|
|
|
|
|
|
@router.post("/maintenance/hermes-update")
|
|
def hermes_update() -> dict:
|
|
return maintenance.hermes_update_job()
|
|
|
|
|
|
@router.post("/maintenance/update-all")
|
|
def update_all(body: SudoReq) -> dict:
|
|
return maintenance.update_all_job(body.sudo_password)
|
|
|
|
|
|
@router.post("/maintenance/reboot")
|
|
def reboot(body: SudoReq) -> dict:
|
|
return maintenance.reboot(body.sudo_password)
|
|
|
|
|
|
@router.post("/maintenance/restart")
|
|
def restart(body: RestartReq) -> dict:
|
|
return maintenance.restart_service(body.service, body.sudo_password)
|
|
|
|
|
|
@router.get("/maintenance/logs")
|
|
def logs(service: str, lines: int = 200, x_sudo_password: str | None = Header(None)) -> dict:
|
|
return maintenance.logs(service, lines, x_sudo_password)
|