102 lines
3.8 KiB
Python
102 lines
3.8 KiB
Python
from fastapi import APIRouter, Depends, Form, HTTPException, Request
|
|
|
|
from ..constants import FLAG_ON
|
|
from ..models import STATE_NAMES
|
|
from ..routing import DIRECTION_DOWN, DIRECTION_UP, Route, RouteError, RouteInput, RouteService
|
|
from ..users import User
|
|
from .deps import FLASH_ERROR, FLASH_OK, current_user, flash, redirect, render, require_admin
|
|
from .routing_form import KIND_LABELS, form_context, parse_form, to_input
|
|
|
|
router = APIRouter()
|
|
|
|
PATH = "/routing"
|
|
STATUS_INVALID = 422
|
|
MSG_NOT_FOUND = "Regel nicht gefunden."
|
|
|
|
|
|
def _service(request: Request) -> RouteService:
|
|
return request.app.state.runtime.routes
|
|
|
|
|
|
def _list_page(request: Request, user: User, data: RouteInput, status_code: int = 200):
|
|
return render(request, "routing.html", user=user, routes=_service(request).list_all(),
|
|
state_names=STATE_NAMES, kind_labels=KIND_LABELS, status_code=status_code,
|
|
**form_context(request, data))
|
|
|
|
|
|
def _get_or_404(request: Request, route_id: int) -> Route:
|
|
route = _service(request).get(route_id)
|
|
if route is None:
|
|
raise HTTPException(404, MSG_NOT_FOUND)
|
|
return route
|
|
|
|
|
|
@router.get(PATH)
|
|
def routing_page(request: Request, user: User = Depends(current_user)):
|
|
return _list_page(request, user, RouteInput())
|
|
|
|
|
|
@router.post(PATH)
|
|
async def routing_create(request: Request, admin: User = Depends(require_admin)):
|
|
data, errors = parse_form(await request.form())
|
|
try:
|
|
if errors:
|
|
raise RouteError(errors)
|
|
_service(request).create(data, admin.username)
|
|
except RouteError as exc:
|
|
for message in exc.errors:
|
|
flash(request, message, FLASH_ERROR)
|
|
return _list_page(request, admin, data, STATUS_INVALID)
|
|
flash(request, "Regel angelegt.", FLASH_OK)
|
|
return redirect(PATH)
|
|
|
|
|
|
@router.get(PATH + "/{route_id}")
|
|
def routing_edit_page(request: Request, route_id: int, admin: User = Depends(require_admin)):
|
|
route = _get_or_404(request, route_id)
|
|
return render(request, "route_edit.html", user=admin, route=route,
|
|
**form_context(request, to_input(route)))
|
|
|
|
|
|
@router.post(PATH + "/{route_id}")
|
|
async def routing_update(request: Request, route_id: int, admin: User = Depends(require_admin)):
|
|
route = _get_or_404(request, route_id)
|
|
data, errors = parse_form(await request.form())
|
|
try:
|
|
if errors:
|
|
raise RouteError(errors)
|
|
_service(request).update(route_id, data, admin.username)
|
|
except RouteError as exc:
|
|
for message in exc.errors:
|
|
flash(request, message, FLASH_ERROR)
|
|
return render(request, "route_edit.html", user=admin, route=route, status_code=STATUS_INVALID,
|
|
**form_context(request, data))
|
|
flash(request, "Regel gespeichert.", FLASH_OK)
|
|
return redirect(PATH)
|
|
|
|
|
|
@router.post(PATH + "/{route_id}/toggle")
|
|
def routing_toggle(request: Request, route_id: int, enabled: str = Form(FLAG_ON),
|
|
admin: User = Depends(require_admin)):
|
|
if not _service(request).set_enabled(route_id, enabled == FLAG_ON, admin.username):
|
|
flash(request, MSG_NOT_FOUND, FLASH_ERROR)
|
|
return redirect(PATH)
|
|
|
|
|
|
@router.post(PATH + "/{route_id}/move")
|
|
def routing_move(request: Request, route_id: int, direction: str = Form(DIRECTION_UP),
|
|
admin: User = Depends(require_admin)):
|
|
if direction not in (DIRECTION_UP, DIRECTION_DOWN):
|
|
raise HTTPException(400, "Ungültige Richtung")
|
|
_service(request).move(route_id, direction, admin.username)
|
|
return redirect(PATH)
|
|
|
|
|
|
@router.post(PATH + "/{route_id}/delete")
|
|
def routing_delete(request: Request, route_id: int, admin: User = Depends(require_admin)):
|
|
if _service(request).delete(route_id, admin.username):
|
|
flash(request, "Regel gelöscht.", FLASH_OK)
|
|
else:
|
|
flash(request, MSG_NOT_FOUND, FLASH_ERROR)
|
|
return redirect(PATH)
|