62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
"""HTTP boundary for bounded historical monitoring trends."""
|
|
|
|
from typing import Annotated
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
|
|
|
|
from app.identity import require_permission, user_has_permission
|
|
from app.monitoring.dependencies import get_monitoring_service
|
|
from app.monitoring.exceptions import MonitoringScopeNotFound
|
|
from app.monitoring.models import (
|
|
MonitoringTrendQuery,
|
|
MonitoringTrendResponse,
|
|
TrendRange,
|
|
TrendScopeType,
|
|
)
|
|
from app.monitoring.service import MonitoringService
|
|
|
|
|
|
router = APIRouter(prefix="/api/monitoring", tags=["monitoring"])
|
|
MonitoringUser = Annotated[dict, Depends(require_permission("miners.view"))]
|
|
MonitoringServiceDependency = Annotated[
|
|
MonitoringService,
|
|
Depends(get_monitoring_service),
|
|
]
|
|
|
|
|
|
@router.get("/trends", response_model=MonitoringTrendResponse)
|
|
def get_monitoring_trends(
|
|
current_user: MonitoringUser,
|
|
service: MonitoringServiceDependency,
|
|
response: Response,
|
|
scope: Annotated[TrendScopeType, Query()] = "fleet",
|
|
scope_id: Annotated[UUID | None, Query()] = None,
|
|
time_range: Annotated[TrendRange, Query(alias="range")] = "24h",
|
|
) -> MonitoringTrendResponse:
|
|
try:
|
|
query = MonitoringTrendQuery(
|
|
scope=scope, scope_id=scope_id, range=time_range
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)
|
|
) from exc
|
|
can_view_locations = user_has_permission(current_user, "locations.view")
|
|
if query.scope in {"site", "room"} and not can_view_locations:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Location viewing permission is required for this scope",
|
|
)
|
|
response.headers["Cache-Control"] = "no-store"
|
|
try:
|
|
return service.get_trends(
|
|
query,
|
|
include_location_scopes=can_view_locations,
|
|
)
|
|
except MonitoringScopeNotFound as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=str(exc),
|
|
) from exc
|