kcar
abf8d05ca1
- timetable_builder_router: Supabase-primary slot write (POST /timetable/slots),
week_cycle support, GET /slots reads from Supabase, materialize-periods endpoint,
rebuild-neo4j endpoint, sync-lessons endpoint (Track B: TaughtLesson Neo4j nodes),
_sync_teacher_timetables_to_neo4j and _sync_taught_lessons_to_neo4j helpers
- classes_router: GET /{class_id} enriched with profiles + enrollment_requests,
GET /school/students for admin search, PATCH /enrollment-requests/{id} approve/reject
- taught_lessons_router: GET /student/lessons student week view with enrichment
- school_router: academic_periods sync, day-type management
- platform_admin_router + platform_admin: POST /admin/reset and /admin/seed endpoints
- invitations_router: teacher invite scaffolding
- reset_environment + seed_environment: idempotent dev environment scripts
- graph_tree_router: Supabase-first institute resolution
- provisioning_service: neo4j_private_db_name column support
- main.py + run/routers.py: register new routers
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
"""
|
|
FastAPI dependencies for platform-level admin access.
|
|
|
|
Two tiers:
|
|
require_platform_admin — user must be in admin_profiles
|
|
require_super_admin — user must have is_super_admin=True in admin_profiles
|
|
|
|
Usage:
|
|
@router.get("/admin/schools")
|
|
async def list_all_schools(admin=Depends(require_platform_admin)):
|
|
...
|
|
|
|
@router.post("/admin/provision")
|
|
async def provision(admin=Depends(require_super_admin)):
|
|
...
|
|
"""
|
|
from fastapi import Depends, HTTPException
|
|
from modules.auth.supabase_bearer import SupabaseBearer
|
|
from modules.database.supabase.utils.client import SupabaseServiceRoleClient
|
|
|
|
|
|
def _sb() -> SupabaseServiceRoleClient:
|
|
return SupabaseServiceRoleClient()
|
|
|
|
|
|
async def require_platform_admin(
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> dict:
|
|
"""Require the caller to be a registered platform admin (in admin_profiles)."""
|
|
user_id = credentials.get("sub")
|
|
if not user_id:
|
|
raise HTTPException(status_code=403, detail="Invalid token")
|
|
try:
|
|
sb = _sb()
|
|
result = (
|
|
sb.supabase.table("admin_profiles")
|
|
.select("id,admin_role,is_super_admin")
|
|
.eq("id", user_id)
|
|
.single()
|
|
.execute()
|
|
)
|
|
if not result.data:
|
|
raise HTTPException(status_code=403, detail="Platform admin access required")
|
|
return {**credentials, "admin_profile": result.data}
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
raise HTTPException(status_code=403, detail="Platform admin access required")
|
|
|
|
|
|
async def require_super_admin(
|
|
admin: dict = Depends(require_platform_admin),
|
|
) -> dict:
|
|
"""Require the caller to have is_super_admin=True."""
|
|
if not admin.get("admin_profile", {}).get("is_super_admin"):
|
|
raise HTTPException(status_code=403, detail="Super admin access required")
|
|
return admin
|