Coverage for app\routers\system.py: 57%

1from typing import List, Dict, Any, Optional 

2from datetime import datetime, UTC 

3from fastapi import APIRouter, Depends, HTTPException, Query 

4from sqlalchemy.orm import Session 

5from pydantic import BaseModel 

6import psutil 

7import json 

8 

9from ..db import SessionLocal, SystemSetting, SystemMetric, SystemComponentStatus, User 

10from ..services.health_service import HealthService 

11from ..deps import get_db, get_current_user 

12 

13router = APIRouter() 

14 

15# --- Pydantic Models --- 

16 

17class SystemSettingSchema(BaseModel): 

18 key: str 

19 value: Any 

20 description: Optional[str] = None 

21 category: Optional[str] = "general" 

22 is_encrypted: bool = False 

23 

24class SystemSettingsUpdate(BaseModel): 

25 settings: Dict[str, Any] 

26 

27class ComponentStatus(BaseModel): 

28 name: str 

29 status: str 

30 message: Optional[str] = None 

31 version: Optional[str] = None 

32 uptime: Optional[float] = None 

33 last_checked: datetime 

34 

35class SystemStatusResponse(BaseModel): 

36 overall_status: str 

37 last_updated: datetime 

38 components: List[ComponentStatus] 

39 

40class SystemMetricsResponse(BaseModel): 

41 cpu_usage: float 

42 memory_usage: float 

43 disk_usage: float 

44 active_users: int 

45 uptime: float 

46 

47# --- Endpoints --- 

48 

49@router.get("/settings", response_model=Dict[str, Any]) 

50def get_system_settings(category: Optional[str] = None, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user)): 

51 """Get all system settings as a key-value dictionary.""" 

52 query = db.query(SystemSetting) 

53 if category: 

54 query = query.filter(SystemSetting.category == category) 

55 

56 settings = query.all() 

57 

58 # Transform list of settings into a dictionary 

59 result = {} 

60 for setting in settings: 

61 result[setting.key] = setting.value 

62 

63 return result 

64 

65@router.put("/settings", response_model=Dict[str, Any]) 

66def update_system_settings(settings_update: Dict[str, Any], db: Session = Depends(get_db), current_user: dict = Depends(get_current_user)): 

67 """Update multiple system settings.""" 

68 # Check permissions (superadmin only) 

69 if current_user.get("role") != "superadmin": 

70 raise HTTPException(status_code=403, detail="Only superadmins can update system settings") 

71 

72 updated_settings = {} 

73 

74 for key, value in settings_update.items(): 

75 setting = db.query(SystemSetting).filter(SystemSetting.key == key).first() 

76 if setting: 

77 setting.value = value 

78 setting.updated_at = datetime.now(UTC) 

79 setting.updated_by = current_user.get("id") 

80 else: 

81 # Create new setting if it doesn't exist 

82 # Infer category from key prefix or default to general 

83 category = "general" 

84 if key.startswith("security_"): category = "security" 

85 elif key.startswith("ai_"): category = "ai" 

86 elif key.startswith("email_"): category = "email" 

87 elif key.startswith("api_"): category = "api" 

88 

89 setting = SystemSetting( 

90 key=key, 

91 value=value, 

92 category=category, 

93 updated_by=current_user.get("id") 

94 ) 

95 db.add(setting) 

96 updated_settings[key] = value 

97 

98 db.commit() 

99 return updated_settings 

100 

101class HeartbeatRequest(BaseModel): 

102 component_name: str 

103 status: str = "ok" 

104 message: Optional[str] = None 

105 version: Optional[str] = None 

106 uptime: Optional[float] = None 

107 metadata: Optional[Dict[str, Any]] = None 

108 

109@router.post("/heartbeat") 

110def receive_heartbeat(heartbeat: HeartbeatRequest, db: Session = Depends(get_db)): 

111 """Receive heartbeat from a system component (e.g. School Node).""" 

112 

113 component = db.query(SystemComponentStatus).filter(SystemComponentStatus.component_name == heartbeat.component_name).first() 

114 

115 if not component: 

116 component = SystemComponentStatus( 

117 component_name=heartbeat.component_name, 

118 status=heartbeat.status, 

119 message=heartbeat.message, 

120 version=heartbeat.version, 

121 uptime=heartbeat.uptime, 

122 last_checked_at=datetime.now(UTC) 

123 ) 

124 db.add(component) 

125 else: 

126 component.status = heartbeat.status 

127 component.message = heartbeat.message 

128 component.version = heartbeat.version 

129 component.uptime = heartbeat.uptime 

130 component.last_checked_at = datetime.now(UTC) 

131 

132 db.commit() 

133 return {"status": "received"} 

134 

135@router.get("/metrics", response_model=SystemMetricsResponse) 

136def get_system_metrics(db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): 

137 """Get current system metrics (CPU, RAM, Disk, Users).""" 

138 

139 # Real-time system metrics using psutil 

140 cpu = psutil.cpu_percent(interval=None) 

141 memory = psutil.virtual_memory().percent 

142 disk = psutil.disk_usage('/').percent 

143 

144 # Mock active users count (in a real app, query Redis or DB session table) 

145 # For now, just count active users in DB as a proxy or use a random number for demo 

146 active_users = 42 # Placeholder 

147 

148 # System uptime 

149 boot_time = psutil.boot_time() 

150 uptime = datetime.now().timestamp() - boot_time 

151 

152 return { 

153 "cpu_usage": cpu, 

154 "memory_usage": memory, 

155 "disk_usage": disk, 

156 "active_users": active_users, 

157 "uptime": uptime 

158 } 

159 

160@router.get("/status") 

161async def get_system_status(db: Session = Depends(get_db)): 

162 service = HealthService(db) 

163 ov = await service.get_system_overview() 

164 now = datetime.now(UTC) 

165 comps: List[ComponentStatus] = [] 

166 for c in ov.components: 

167 last_checked = getattr(c, "last_updated", None) or now 

168 comps.append(ComponentStatus( 

169 name=c.name, 

170 status=c.status.value if hasattr(c.status, "value") else c.status, 

171 message=c.message, 

172 version=(c.metadata or {}).get("version"), 

173 uptime=(c.metadata or {}).get("uptime"), 

174 last_checked=last_checked, 

175 )) 

176 return { 

177 "overall_status": ov.overall_status.value if hasattr(ov.overall_status, "value") else ov.overall_status, 

178 "last_updated": (ov.last_updated.isoformat() if hasattr(ov.last_updated, "isoformat") else ov.last_updated), 

179 "components": [ 

180 { 

181 "name": c.name, 

182 "status": c.status, 

183 "message": c.message, 

184 "version": c.version, 

185 "uptime": c.uptime, 

186 "last_checked": c.last_checked.isoformat() if hasattr(c.last_checked, "isoformat") else c.last_checked, 

187 } for c in comps 

188 ], 

189 }