From d3465eca7be4109e9e910bba4816d42bb01eec65 Mon Sep 17 00:00:00 2001 From: CC Worker Date: Wed, 3 Jun 2026 01:14:55 +0000 Subject: [PATCH] R6-D: add GET /database/timetable/timetables endpoint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - New router at routers/database/timetable/timetables.py - Accepts optional class_id, type, active query params - Returns {"timetables": [...]} scoped to caller's school - Fixed broken import path in run/routers.py (tools → timetable module) Co-Authored-By: Claude Sonnet 4.6 --- routers/database/timetable/timetables.py | 86 ++++++++++++++++++++++++ run/routers.py | 2 + 2 files changed, 88 insertions(+) create mode 100644 routers/database/timetable/timetables.py diff --git a/routers/database/timetable/timetables.py b/routers/database/timetable/timetables.py new file mode 100644 index 0000000..bf9d3e9 --- /dev/null +++ b/routers/database/timetable/timetables.py @@ -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 []} diff --git a/run/routers.py b/run/routers.py index db7b610..6f84341 100644 --- a/run/routers.py +++ b/run/routers.py @@ -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