Coverage for app\routers\analytics.py: 31%

1from typing import List, Optional 

2from datetime import datetime, timedelta, UTC 

3from fastapi import APIRouter, Depends, Query, HTTPException 

4from sqlalchemy.orm import Session 

5from sqlalchemy import func, desc 

6from ..db import SessionLocal, AnalyticsEvent, User, SystemMetric 

7from ..schemas import analytics as schemas 

8from ..deps import get_db 

9 

10router = APIRouter() 

11 

12@router.get("/summary", response_model=schemas.AnalyticsSummary) 

13def get_analytics_summary( 

14 time_range: str = Query("24h"), 

15 db: Session = Depends(get_db) 

16): 

17 # Calculate time delta 

18 now = datetime.now(UTC) 

19 if time_range == "24h": 

20 start_time = now - timedelta(hours=24) 

21 elif time_range == "7d": 

22 start_time = now - timedelta(days=7) 

23 elif time_range == "30d": 

24 start_time = now - timedelta(days=30) 

25 else: 

26 start_time = now - timedelta(hours=24) 

27 

28 # Base query 

29 query = db.query(AnalyticsEvent).filter(AnalyticsEvent.created_at >= start_time) 

30 

31 total_events = query.count() 

32 active_users = query.filter(AnalyticsEvent.user_id.isnot(None)).distinct(AnalyticsEvent.user_id).count() 

33 

34 error_count = query.filter(AnalyticsEvent.status == "error").count() 

35 error_rate = (error_count / total_events) if total_events > 0 else 0.0 

36 

37 avg_duration = db.query(func.avg(AnalyticsEvent.duration)).filter(AnalyticsEvent.created_at >= start_time).scalar() or 0.0 

38 

39 # Top actions 

40 top_actions_res = db.query( 

41 AnalyticsEvent.action, func.count(AnalyticsEvent.id).label("count") 

42 ).filter( 

43 AnalyticsEvent.created_at >= start_time 

44 ).group_by(AnalyticsEvent.action).order_by(desc("count")).limit(5).all() 

45 

46 top_actions = [{"action": r[0], "count": r[1]} for r in top_actions_res] 

47 

48 # Categories 

49 categories = { 

50 "authentication": query.filter(AnalyticsEvent.event_category == "authentication").count(), 

51 "data_operation": query.filter(AnalyticsEvent.event_category == "data_operation").count(), 

52 "system_health": query.filter(AnalyticsEvent.event_category == "system_health").count(), 

53 } 

54 

55 return { 

56 "time_range": time_range, 

57 "total_events": total_events, 

58 "active_users": active_users, 

59 "error_rate": error_rate, 

60 "avg_response_time": avg_duration, 

61 "top_actions": top_actions, 

62 "event_categories": categories 

63 } 

64 

65@router.get("/events", response_model=schemas.EventsResponse) 

66def get_events( 

67 time_range: str = Query("24h"), 

68 event_type: Optional[str] = None, 

69 status: Optional[str] = None, 

70 limit: int = 100, 

71 db: Session = Depends(get_db) 

72): 

73 now = datetime.now(UTC) 

74 if time_range == "24h": 

75 start_time = now - timedelta(hours=24) 

76 else: 

77 start_time = now - timedelta(days=7) # Simplification 

78 

79 query = db.query(AnalyticsEvent).filter(AnalyticsEvent.created_at >= start_time) 

80 

81 if event_type: 

82 query = query.filter(AnalyticsEvent.event_type == event_type) 

83 if status: 

84 query = query.filter(AnalyticsEvent.status == status) 

85 

86 events = query.order_by(desc(AnalyticsEvent.created_at)).limit(limit).all() 

87 return {"events": events} 

88 

89@router.post("/events", response_model=schemas.AnalyticsEvent) 

90def create_event( 

91 event: schemas.AnalyticsEventCreate, 

92 db: Session = Depends(get_db) 

93): 

94 db_event = AnalyticsEvent( 

95 event_type=event.event_type, 

96 event_category=event.event_category, 

97 action=event.action, 

98 status=event.status, 

99 duration=event.duration, 

100 properties=event.properties, 

101 created_at=datetime.now(UTC) 

102 ) 

103 db.add(db_event) 

104 db.commit() 

105 db.refresh(db_event) 

106 return db_event 

107 

108@router.get("/dashboard/users", response_model=schemas.UserAnalytics) 

109def get_user_analytics(time_range: str = Query("24h"), db: Session = Depends(get_db)): 

110 # Mock implementation for now as we don't have full history tracking for users yet 

111 # In a real system, we would query a UserHistory table or aggregate from logs 

112 

113 total_active = db.query(User).filter(User.status == "active").count() 

114 total_inactive = db.query(User).filter(User.status == "inactive").count() 

115 total_suspended = db.query(User).filter(User.status == "suspended").count() 

116 

117 # Roles 

118 roles_res = db.query(User.role, func.count(User.id)).group_by(User.role).all() 

119 users_by_role = [{"role": str(r[0]), "count": r[1]} for r in roles_res] 

120 

121 return { 

122 "user_registrations": [], # TODO: Implement real aggregation 

123 "users_by_role": users_by_role, 

124 "login_activity": { 

125 "successful_logins": 0, 

126 "failed_logins": 0, 

127 "total_attempts": 0 

128 }, 

129 "total_active_users": total_active, 

130 "total_inactive_users": total_inactive, 

131 "total_suspended_users": total_suspended 

132 } 

133 

134@router.get("/dashboard/system", response_model=schemas.SystemAnalytics) 

135def get_system_analytics(time_range: str = Query("24h"), db: Session = Depends(get_db)): 

136 # Get recent system metrics 

137 metrics = db.query(SystemMetric).order_by(desc(SystemMetric.created_at)).limit(50).all() 

138 

139 return { 

140 "system_metrics": [], # TODO: Map from SystemMetric 

141 "api_statistics": { 

142 "total_calls": 0, 

143 "avg_response_time": 0, 

144 "error_rate": 0 

145 }, 

146 "system_health": { 

147 "errors": 0, 

148 "database_operations": 0, 

149 "uptime_percentage": 99.9 

150 }, 

151 "performance_metrics": { 

152 "slow_queries": 0, 

153 "peak_load_time": "0ms" 

154 } 

155 }