Spaces:
Sleeping
Sleeping
File size: 2,651 Bytes
d813a22 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | """FastAPI backend for PulseFabric Monitor dashboard.
Serves the same 4 REST endpoints that the Angular frontend expects,
loading data from db.json at startup so the data source is unchanged.
"""
import json
from pathlib import Path
from typing import Literal
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
# ---------------------------------------------------------------------------
# Pydantic models (mirrors frontend api.models.ts)
# ---------------------------------------------------------------------------
class KpiDto(BaseModel):
id: str
title: str
value: str
unit: str | None = None
trend: Literal["up", "down", "stable"]
trendLabel: str
icon: str
class TrafficDto(BaseModel):
time: str
inbound: float
outbound: float
latency: float
class IncidentDto(BaseModel):
region: str
critical: int
warning: int
info: int
class AlertDto(BaseModel):
id: int
timestamp: str
severity: Literal["critical", "warning", "info"]
message: str
source: str
# ---------------------------------------------------------------------------
# Load data from db.json at startup
# ---------------------------------------------------------------------------
_DB_PATH = Path(__file__).parent / "db.json"
with _DB_PATH.open(encoding="utf-8") as _f:
_DB: dict = json.load(_f)
# ---------------------------------------------------------------------------
# App
# ---------------------------------------------------------------------------
app = FastAPI(
title="PulseFabric Monitor API",
description="Backend for the PulseFabric network operations monitoring dashboard.",
version="1.0.0",
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["GET"],
allow_headers=["*"],
)
# ---------------------------------------------------------------------------
# Routes
# ---------------------------------------------------------------------------
@app.get("/kpis", response_model=list[KpiDto])
def get_kpis() -> list[dict]:
"""Return all KPI cards."""
return _DB["kpis"]
@app.get("/traffic", response_model=list[TrafficDto])
def get_traffic() -> list[dict]:
"""Return 24-hour traffic and latency data."""
return _DB["traffic"]
@app.get("/incidents", response_model=list[IncidentDto])
def get_incidents() -> list[dict]:
"""Return incident severity counts by region."""
return _DB["incidents"]
@app.get("/alerts", response_model=list[AlertDto])
def get_alerts() -> list[dict]:
"""Return recent alert feed."""
return _DB["alerts"]
|