Coverage for src/cstlcore/admin/router.py: 60%
81 statements
« prev ^ index » next coverage.py v7.9.1, created at 2026-02-19 12:46 +0000
1from typing import Optional
3from fastapi import APIRouter, Depends, HTTPException, Query
4from fastapi.security import OAuth2PasswordBearer
5from sqlalchemy.orm import Session
7from cstlcore.auth.dependencies import require_admin_user
8from cstlcore.database.dependencies import get_session
9from cstlcore.memberships.dependencies import require_read_access
10from cstlcore.users.models import User
11from cstlcore.settings import settings
12from cstlcore.admin.services import (
13 get_admin_configuration_info,
14 get_dashboard_stats,
15 get_dashboard_activity,
16 get_dashboard_alerts,
17 get_users_list,
18 toggle_user_status,
19 get_constellations_list,
20 get_analytics_data,
21 get_permissions_list,
22 update_permission_role,
23 revoke_permission,
24 get_constellation_graph_metrics,
25 get_all_constellations_graph_metrics,
26)
27from cstlcore.admin.models import (
28 UpdateRoleRequest,
29 GraphMetricsResponse,
30 GraphMetricsBulkResponse,
31)
34router = APIRouter()
36# OAuth2 scheme for token extraction
37oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")
40@router.get("/admin")
41async def admin_status(current_user: User = Depends(require_admin_user)):
42 return {"status": "ok"}
45@router.get("/admin/users") #, response_model=list[UserPublic])
46async def list_users(
47 current_user: User = Depends(require_admin_user),
48 session: Session = Depends(get_session),
49):
50 users = session.query(User).all()
51 return users
54@router.get("/admin/users/{user_id}") #, response_model=UserPublic)
55async def get_user_admin(
56 user_id: str,
57 current_user: User = Depends(require_admin_user),
58 session: Session = Depends(get_session),
59):
60 user = session.query(User).filter(User.id == user_id).first()
61 if not user:
62 raise HTTPException(status_code=404, detail="User not found")
63 return user
67@router.get("/admin/check-config")
68async def check_admin_config(_current_user: User = Depends(require_admin_user)):
69 """
70 Check the current admin configuration.
71 Public endpoint for debugging configuration.
72 """
73 return get_admin_configuration_info()
76@router.get("/admin/is-user-admin/{user_email}")
77async def is_user_admin(user_email: str):
78 """
79 Check if a specific email has admin privileges.
80 Public endpoint for debugging.
81 """
82 is_admin = settings.admin.is_admin_email(user_email)
83 return {
84 "email": user_email,
85 "is_admin": is_admin
86 }
89# Todo we will rename all of this to backoffice instead of admin/dashboard
92#################################################################################################################################################################################
94# Dashboard Endpoints
95@router.get("/dashboard/stats") #, response_model=DashboardStatsResponse)
96async def get_dashboard_stats_endpoint(
97 _current_user: User = Depends(require_admin_user),
98 session: Session = Depends(get_session)
99):
100 """Get overview statistics for the admin dashboard"""
101 return get_dashboard_stats(session)
104@router.get("/dashboard/activity") #, response_model=DashboardActivityResponse)
105async def get_dashboard_activity_endpoint(
106 _current_user: User = Depends(require_admin_user),
107 session: Session = Depends(get_session)
108):
109 """Get recent platform activity"""
110 return get_dashboard_activity(session)
113@router.get("/dashboard/alerts") #, response_model=DashboardAlertsResponse)
114async def get_dashboard_alerts_endpoint(
115 _current_user: User = Depends(require_admin_user),
116 session: Session = Depends(get_session)
117):
118 """Get system alerts and notifications"""
119 return get_dashboard_alerts(session)
121#################################################################################################################################################################################
124# I had bad requests
125# /dashboard/users?page=1&limit=50
127# User Management Endpoints
128@router.get("/dashboard/users") #, response_model=UsersListResponse)
129async def get_users_endpoint(
130 current_user: User = Depends(require_admin_user),
131 session: Session = Depends(get_session),
132 search: Optional[str] = Query(None),
133 status: Optional[str] = Query(None),
134 verified: Optional[bool] = Query(None)
135):
136 """Get all users with filtering"""
137 return get_users_list(session, search, status, verified)
140@router.post("/dashboard/users/{user_id}/toggle-status") #, response_model=UserToggleStatusResponse)
141async def toggle_user_status_endpoint(
142 user_id: str,
143 current_user: User = Depends(require_admin_user),
144 session: Session = Depends(get_session)
145):
146 """Toggle user active/inactive status"""
147 result = toggle_user_status(session, user_id)
148 if not result["success"]:
149 raise HTTPException(status_code=404, detail=result["message"])
150 return result
153@router.post("/dashboard/users/{user_id}/send-verification") #, response_model=UserSendVerificationResponse)
154async def send_verification_email_endpoint(
155 user_id: str,
156 current_user: User = Depends(require_admin_user),
157 session: Session = Depends(get_session)
158):
159 """Send verification email to user"""
160 # TODO: Implement actual email sending
161 return {
162 "success": True,
163 "message": "Verification email sent successfully"
164 }
167#################################################################################################################################################################################
169# Constellation Management Endpoints
170@router.get("/dashboard/constellations") #, response_model=ConstellationsListResponse)
171async def get_constellations_endpoint(
172 current_user: User = Depends(require_admin_user),
173 session: Session = Depends(get_session),
174 search: Optional[str] = Query(None),
175 sort_by: Optional[str] = Query(None)
176):
177 """Get all constellations with filtering"""
178 return get_constellations_list(session, search, sort_by)
182# Permissions Management Endpoints
183@router.get("/dashboard/permissions")
184async def get_permissions_endpoint(
185 current_user: User = Depends(require_admin_user),
186 session: Session = Depends(get_session),
187 search: Optional[str] = Query(None),
188 role: Optional[str] = Query(None)
189):
190 """Get all user-constellation permissions"""
191 return get_permissions_list(session, search, role)
194# TODO
195# /dashboard/permissions/c14222ea-f41f-4579-8339-8c4b6c02d0e0_9942ff0d-2c1c-48d1-a1d5-aa8cae50803a/role
196@router.put("/dashboard/permissions/{permission_id}/role") #, response_model=UpdateRoleResponse)
197async def update_permission_role_endpoint(
198 permission_id: str,
199 role_data: UpdateRoleRequest,
200 current_user: User = Depends(require_admin_user),
201 session: Session = Depends(get_session)
202):
203 """Update user role for a constellation"""
204 result = update_permission_role(session, permission_id, role_data.role)
205 if not result["success"]:
206 raise HTTPException(status_code=400, detail=result["message"])
207 return result
210@router.delete("/dashboard/permissions/{permission_id}") #, response_model=RevokePermissionResponse)
211async def revoke_permission_endpoint(
212 permission_id: str,
213 current_user: User = Depends(require_admin_user),
214 session: Session = Depends(get_session)
215):
216 """Revoke user permission from a constellation"""
217 result = revoke_permission(session, permission_id)
218 if not result["success"]:
219 raise HTTPException(status_code=400, detail=result["message"])
220 return result
223# Analytics Endpoints
224@router.get("/dashboard/analytics") #, response_model=AnalyticsResponse)
225async def get_analytics_endpoint(
226 current_user: User = Depends(require_admin_user),
227 session: Session = Depends(get_session),
228 period: str = Query("30d", pattern="^(7d|30d|90d|1y)$")
229):
230 """Get comprehensive analytics data"""
231 return get_analytics_data(session, period)
234#################################################################################################################################################################################
236# Graph Theory Metrics Endpoints
237@router.get(
238 "/constellations/{constellation_id}/graph-metrics",
239 response_model=GraphMetricsResponse,
240 tags=["Graph Metrics"],
241)
242async def get_constellation_graph_metrics_endpoint(
243 constellation_id: str,
244 constellation=Depends(require_read_access),
245 session: Session = Depends(get_session),
246 token: str = Depends(oauth2_scheme),
247):
248 """
249 Get graph theory metrics for a constellation.
251 Accessible to any user with READ access to the constellation.
253 Computes:
254 - Element count (N): Number of narrative elements (nodes)
255 - Relation count (M): Number of unique undirected edges
256 - Graph density (D): D = 2M / N(N-1)
257 - Average degree (k): k = 2M / N
258 - Narrative interpretation based on density thresholds
259 """
260 return await get_constellation_graph_metrics(session, constellation_id, token)
263@router.get(
264 "/dashboard/analytics/graph-metrics",
265 response_model=GraphMetricsBulkResponse,
266 tags=["Graph Metrics"],
267)
268async def get_all_graph_metrics_endpoint(
269 _current_user: User = Depends(require_admin_user),
270 session: Session = Depends(get_session),
271 token: str = Depends(oauth2_scheme),
272):
273 """
274 Get graph theory metrics for all constellations (bulk).
276 Returns aggregated metrics for project comparison and
277 structural progression tracking across the platform.
278 """
279 return await get_all_constellations_graph_metrics(session, token)