71 lines
3.4 KiB
Python
71 lines
3.4 KiB
Python
import json
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
|
|
from app.api.deps import db_session, get_current_user
|
|
from app.api.routes_routers import require_csrf
|
|
from app.models.user import User
|
|
from app.models.dashboard import Dashboard, DashboardPanel, Permission
|
|
from app.models.router import Router
|
|
from app.schemas.dashboard import DashboardCreate, DashboardOut, PanelCreate, PanelOut
|
|
|
|
router = APIRouter()
|
|
|
|
@router.get("", response_model=list[DashboardOut])
|
|
async def list_dashboards(user: User = Depends(get_current_user), session: AsyncSession = Depends(db_session)):
|
|
res = await session.execute(select(Dashboard).where(Dashboard.owner_user_id == user.id))
|
|
rows = res.scalars().all()
|
|
return [DashboardOut(id=d.id, name=d.name, is_shared=d.is_shared) for d in rows]
|
|
|
|
@router.post("", response_model=DashboardOut)
|
|
async def create_dashboard(payload: DashboardCreate, request: Request, user: User = Depends(get_current_user), session: AsyncSession = Depends(db_session)):
|
|
require_csrf(request)
|
|
d = Dashboard(owner_user_id=user.id, name=payload.name, is_shared=payload.is_shared)
|
|
session.add(d)
|
|
await session.commit()
|
|
await session.refresh(d)
|
|
return DashboardOut(id=d.id, name=d.name, is_shared=d.is_shared)
|
|
|
|
@router.get("/{dashboard_id}")
|
|
async def get_dashboard(dashboard_id: int, user: User = Depends(get_current_user), session: AsyncSession = Depends(db_session)):
|
|
res = await session.execute(select(Dashboard).where(Dashboard.id == dashboard_id))
|
|
d = res.scalar_one_or_none()
|
|
if not d or d.owner_user_id != user.id:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
pres = await session.execute(select(DashboardPanel).where(DashboardPanel.dashboard_id == d.id))
|
|
panels = pres.scalars().all()
|
|
out_panels = []
|
|
for p in panels:
|
|
out_panels.append(PanelOut(
|
|
id=p.id, dashboard_id=p.dashboard_id, title=p.title,
|
|
router_id=p.router_id, config=json.loads(p.config_json or "{}")
|
|
).model_dump())
|
|
return {"dashboard": DashboardOut(id=d.id, name=d.name, is_shared=d.is_shared).model_dump(), "panels": out_panels}
|
|
|
|
@router.post("/{dashboard_id}/panels", response_model=PanelOut)
|
|
async def create_panel(dashboard_id: int, payload: PanelCreate, request: Request, user: User = Depends(get_current_user), session: AsyncSession = Depends(db_session)):
|
|
require_csrf(request)
|
|
# dashboard ownership
|
|
res = await session.execute(select(Dashboard).where(Dashboard.id == dashboard_id))
|
|
d = res.scalar_one_or_none()
|
|
if not d or d.owner_user_id != user.id:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
|
|
# check permission to router
|
|
if user.role != "admin":
|
|
pres = await session.execute(select(Permission).where(Permission.user_id == user.id, Permission.router_id == payload.router_id, Permission.can_view == True)) # noqa
|
|
if not pres.scalar_one_or_none():
|
|
raise HTTPException(status_code=403, detail="No access to router")
|
|
|
|
p = DashboardPanel(
|
|
dashboard_id=dashboard_id,
|
|
title=payload.title,
|
|
router_id=payload.router_id,
|
|
config_json=json.dumps(payload.config or {}),
|
|
)
|
|
session.add(p)
|
|
await session.commit()
|
|
await session.refresh(p)
|
|
return PanelOut(id=p.id, dashboard_id=p.dashboard_id, title=p.title, router_id=p.router_id, config=payload.config)
|