Compare commits
No commits in common. "3711b52ea4e7d575b8067839cc81f8ed5db39891" and "f203f376e9aad3480b825182bb0b241609c5ac17" have entirely different histories.
3711b52ea4
...
f203f376e9
@ -1,86 +0,0 @@
|
|||||||
"""
|
|
||||||
GET /database/timetable/timetables
|
|
||||||
Optional filters: class_id, type, active
|
|
||||||
Returns {"timetables": [...]} for the caller's school.
|
|
||||||
"""
|
|
||||||
import os
|
|
||||||
from typing import Any, Dict, List, Optional
|
|
||||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
||||||
from pydantic import BaseModel
|
|
||||||
from modules.logger_tool import initialise_logger
|
|
||||||
from modules.auth.supabase_bearer import SupabaseBearer
|
|
||||||
from modules.database.supabase.utils.client import SupabaseServiceRoleClient
|
|
||||||
|
|
||||||
ADMIN_TYPES = ("school_admin", "department_head")
|
|
||||||
|
|
||||||
logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), 'default', True)
|
|
||||||
router = APIRouter()
|
|
||||||
|
|
||||||
|
|
||||||
class TimetableResponse(BaseModel):
|
|
||||||
timetables: List[Dict[str, Any]]
|
|
||||||
|
|
||||||
|
|
||||||
def _sb() -> SupabaseServiceRoleClient:
|
|
||||||
return SupabaseServiceRoleClient()
|
|
||||||
|
|
||||||
|
|
||||||
def _require_institute(user_id: str) -> Optional[str]:
|
|
||||||
try:
|
|
||||||
sb = _sb()
|
|
||||||
p = sb.supabase.table("profiles").select("school_id").eq("id", user_id).single().execute()
|
|
||||||
return str((p.data or {}).get("school_id") or "")
|
|
||||||
except Exception:
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
|
||||||
def _is_admin(user_id: str, institute_id: str) -> bool:
|
|
||||||
try:
|
|
||||||
sb = _sb()
|
|
||||||
r = (
|
|
||||||
sb.supabase.table("institute_memberships")
|
|
||||||
.select("role")
|
|
||||||
.eq("profile_id", user_id)
|
|
||||||
.eq("institute_id", institute_id)
|
|
||||||
.in_("role", list(ADMIN_TYPES))
|
|
||||||
.limit(1)
|
|
||||||
.execute()
|
|
||||||
)
|
|
||||||
return bool(r.data)
|
|
||||||
except Exception:
|
|
||||||
return False
|
|
||||||
|
|
||||||
|
|
||||||
@router.get("", response_model=TimetableResponse)
|
|
||||||
async def list_timetables(
|
|
||||||
class_id: Optional[str] = Query(None),
|
|
||||||
type: Optional[str] = Query(None),
|
|
||||||
active: Optional[bool] = Query(None),
|
|
||||||
credentials: dict = Depends(SupabaseBearer()),
|
|
||||||
) -> Dict[str, Any]:
|
|
||||||
user_id = credentials.get("sub", "")
|
|
||||||
institute_id = _require_institute(user_id)
|
|
||||||
|
|
||||||
if not institute_id:
|
|
||||||
return {"timetables": []}
|
|
||||||
|
|
||||||
sb = _sb()
|
|
||||||
|
|
||||||
if not _is_admin(user_id, institute_id):
|
|
||||||
return {"timetables": []}
|
|
||||||
|
|
||||||
q = (
|
|
||||||
sb.supabase.table("school_timetables")
|
|
||||||
.select("*")
|
|
||||||
.eq("institute_id", institute_id)
|
|
||||||
)
|
|
||||||
|
|
||||||
if class_id:
|
|
||||||
q = q.eq("class_id", class_id)
|
|
||||||
if type:
|
|
||||||
q = q.eq("type", type)
|
|
||||||
if active is not None:
|
|
||||||
q = q.eq("is_active", active)
|
|
||||||
|
|
||||||
res = q.order("created_at", desc=True).execute()
|
|
||||||
return {"timetables": res.data or []}
|
|
||||||
@ -31,9 +31,12 @@ def _resolve_institute_id(user_id: str) -> Optional[str]:
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
def _require_institute(user_id: str) -> Optional[str]:
|
def _require_institute(user_id: str) -> str:
|
||||||
"""Return institute_id, or None if the user has no school membership."""
|
"""Return institute_id or raise 400."""
|
||||||
return _resolve_institute_id(user_id)
|
institute_id = _resolve_institute_id(user_id)
|
||||||
|
if not institute_id:
|
||||||
|
raise HTTPException(status_code=400, detail="User is not linked to a school")
|
||||||
|
return institute_id
|
||||||
|
|
||||||
|
|
||||||
def _is_school_admin(user_id: str, institute_id: str) -> bool:
|
def _is_school_admin(user_id: str, institute_id: str) -> bool:
|
||||||
@ -102,8 +105,6 @@ async def list_classes(
|
|||||||
) -> Dict[str, Any]:
|
) -> Dict[str, Any]:
|
||||||
user_id = credentials.get("sub", "")
|
user_id = credentials.get("sub", "")
|
||||||
institute_id = _require_institute(user_id)
|
institute_id = _require_institute(user_id)
|
||||||
if not institute_id:
|
|
||||||
return {"classes": [], "total": 0}
|
|
||||||
sb = _sb()
|
sb = _sb()
|
||||||
|
|
||||||
q = sb.supabase.table("classes").select("*", count="exact").eq("institute_id", institute_id)
|
q = sb.supabase.table("classes").select("*", count="exact").eq("institute_id", institute_id)
|
||||||
@ -168,8 +169,6 @@ async def my_teaching_classes(
|
|||||||
) -> Dict[str, Any]:
|
) -> Dict[str, Any]:
|
||||||
user_id = credentials.get("sub", "")
|
user_id = credentials.get("sub", "")
|
||||||
institute_id = _require_institute(user_id)
|
institute_id = _require_institute(user_id)
|
||||||
if not institute_id:
|
|
||||||
return {"classes": []}
|
|
||||||
sb = _sb()
|
sb = _sb()
|
||||||
|
|
||||||
assigned = (
|
assigned = (
|
||||||
@ -205,8 +204,6 @@ async def my_student_classes(
|
|||||||
) -> Dict[str, Any]:
|
) -> Dict[str, Any]:
|
||||||
user_id = credentials.get("sub", "")
|
user_id = credentials.get("sub", "")
|
||||||
institute_id = _require_institute(user_id)
|
institute_id = _require_institute(user_id)
|
||||||
if not institute_id:
|
|
||||||
return {"classes": []}
|
|
||||||
sb = _sb()
|
sb = _sb()
|
||||||
|
|
||||||
enrolled = (
|
enrolled = (
|
||||||
@ -240,8 +237,6 @@ async def list_school_students(
|
|||||||
"""List all students in the caller's school. Used by admin to add students to a class."""
|
"""List all students in the caller's school. Used by admin to add students to a class."""
|
||||||
user_id = credentials.get("sub", "")
|
user_id = credentials.get("sub", "")
|
||||||
institute_id = _require_institute(user_id)
|
institute_id = _require_institute(user_id)
|
||||||
if not institute_id:
|
|
||||||
return {"students": []}
|
|
||||||
sb = _sb()
|
sb = _sb()
|
||||||
members = (
|
members = (
|
||||||
sb.supabase.table("institute_memberships")
|
sb.supabase.table("institute_memberships")
|
||||||
|
|||||||
@ -58,9 +58,7 @@ def register_routes(app: FastAPI):
|
|||||||
app.include_router(entity_init.router, prefix="/database/entity", tags=["Entity"])
|
app.include_router(entity_init.router, prefix="/database/entity", tags=["Entity"])
|
||||||
app.include_router(calendar.router, prefix="/database/calendar", tags=["Calendar"])
|
app.include_router(calendar.router, prefix="/database/calendar", tags=["Calendar"])
|
||||||
app.include_router(schools.router, prefix="/database/schools", tags=["Schools"])
|
app.include_router(schools.router, prefix="/database/schools", tags=["Schools"])
|
||||||
from routers.database.timetable.timetables import router as timetable_router
|
|
||||||
app.include_router(timetables.router, prefix="/database/timetables", tags=["Timetables"])
|
app.include_router(timetables.router, prefix="/database/timetables", tags=["Timetables"])
|
||||||
app.include_router(timetable_router, prefix="/database/timetable/timetables", tags=["Timetables"])
|
|
||||||
app.include_router(curriculum.router, prefix="/database/curriculum", tags=["Curriculum"])
|
app.include_router(curriculum.router, prefix="/database/curriculum", tags=["Curriculum"])
|
||||||
|
|
||||||
# Navigation Routes
|
# Navigation Routes
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user