R6-D: add GET /database/timetable/timetables endpoint
- 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 <[email protected]>
This commit is contained in:
co-authored by
Claude Sonnet 4.6
parent
9de949d212
commit
d3465eca7b
@@ -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 []}
|
||||
Reference in New Issue
Block a user