Merge fix/r6-timetable-endpoint: R6-D timetable endpoint + R6-E classes fix
Some checks failed
api-ci-deploy / test-build-deploy (push) Has been cancelled

- GET /database/timetable/timetables with optional filters (R6-D)
- Return empty collections instead of 400 when user has no school (R6-E)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
CC Worker 2026-06-03 01:15:00 +00:00
commit 3711b52ea4
3 changed files with 99 additions and 6 deletions

View File

@ -0,0 +1,86 @@
"""
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 []}

View File

@ -31,12 +31,9 @@ def _resolve_institute_id(user_id: str) -> Optional[str]:
return None
def _require_institute(user_id: str) -> str:
"""Return institute_id or raise 400."""
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 _require_institute(user_id: str) -> Optional[str]:
"""Return institute_id, or None if the user has no school membership."""
return _resolve_institute_id(user_id)
def _is_school_admin(user_id: str, institute_id: str) -> bool:
@ -105,6 +102,8 @@ async def list_classes(
) -> Dict[str, Any]:
user_id = credentials.get("sub", "")
institute_id = _require_institute(user_id)
if not institute_id:
return {"classes": [], "total": 0}
sb = _sb()
q = sb.supabase.table("classes").select("*", count="exact").eq("institute_id", institute_id)
@ -169,6 +168,8 @@ async def my_teaching_classes(
) -> Dict[str, Any]:
user_id = credentials.get("sub", "")
institute_id = _require_institute(user_id)
if not institute_id:
return {"classes": []}
sb = _sb()
assigned = (
@ -204,6 +205,8 @@ async def my_student_classes(
) -> Dict[str, Any]:
user_id = credentials.get("sub", "")
institute_id = _require_institute(user_id)
if not institute_id:
return {"classes": []}
sb = _sb()
enrolled = (
@ -237,6 +240,8 @@ async def list_school_students(
"""List all students in the caller's school. Used by admin to add students to a class."""
user_id = credentials.get("sub", "")
institute_id = _require_institute(user_id)
if not institute_id:
return {"students": []}
sb = _sb()
members = (
sb.supabase.table("institute_memberships")

View File

@ -58,7 +58,9 @@ def register_routes(app: FastAPI):
app.include_router(entity_init.router, prefix="/database/entity", tags=["Entity"])
app.include_router(calendar.router, prefix="/database/calendar", tags=["Calendar"])
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(timetable_router, prefix="/database/timetable/timetables", tags=["Timetables"])
app.include_router(curriculum.router, prefix="/database/curriculum", tags=["Curriculum"])
# Navigation Routes