"""
Dashboard API Endpoints
GET /api/v1/dashboard/at - AT dashboard stats and active cases
"""
from typing import List, Optional, Dict
from datetime import datetime, timedelta
from fastapi import APIRouter, Depends, HTTPException, Query, status, Request
from sqlalchemy.orm import Session
from sqlalchemy import and_, or_, func
from pydantic import BaseModel, Field
from app.database import get_db
from app.models import Athlete, Case, Event, School, User, Milestone
from app.schemas import EventContent, AthleteSchema, EventSchema
from app.utils.id_generator import generate_id
from app.auth import get_current_user, require_role, require_coach_or_ad, require_ad
router = APIRouter(tags=["dashboard"])
============== SCHEMAS ==============
class CaseSummary(BaseModel):
"""Case summary for dashboard list"""
athlete_id: str
athlete_name: str
initials: str
sport: str
injury: str
injury_short: str
status: str
status_label: str
avatar_color: str
class ActivityItem(BaseModel):
"""Recent activity item"""
type: str # cleared, out, modified
text: str
time_ago: str
class DashboardStats(BaseModel):
"""Dashboard statistics"""
full_count: int
modified_count: int
out_count: int
class CoachDashboardStats(BaseModel):
"""Coach dashboard statistics"""
out: int
modified: int
cleared: int
class AthleteSummary(BaseModel):
"""Athlete summary for team roster"""
id: str
name: str
grade: int
position: Optional[str] = None
current_status: str
status_label: str
case_title: Optional[str] = None
avatar_color: str
class CoachDashboardResponse(BaseModel):
"""Coach Dashboard response"""
sport: str
stats: CoachDashboardStats
teams: Dict[str, List[AthleteSummary]]
recent_activity: List[ActivityItem]
class SportSummary(BaseModel):
"""Sport summary for AD view"""
name: str
athletes: int
out: int
modified: int
class ADDashboardResponse(BaseModel):
"""AD Dashboard response"""
school_id: str
total_athletes: int
active_cases: int
out_today: int
modified: int
sports: List[SportSummary]
recent_activity: List[ActivityItem]
class ATDashboardResponse(BaseModel):
"""AT Dashboard response"""
stats: DashboardStats
active_cases: List[CaseSummary]
recent_activity: List[ActivityItem]
============== PARENT DASHBOARD SCHEMAS ==============
class ParentChildSummary(BaseModel):
"""Child athlete summary for parent dashboard"""
id: str
first_name: str
last_name: str
sport: str
team: Optional[str] = None
grade: int
has_active_case: bool
current_status: str
initials: str
avatar_color: str
class ParentChildrenResponse(BaseModel):
"""Parent dashboard children list"""
children: List[ParentChildSummary]
class ParentMilestoneSummary(BaseModel):
"""Milestone summary for parent dashboard"""
id: str
title: str
status: str
target_date: Optional[datetime] = None
achieved_at: Optional[datetime] = None
class ParentTimelineItem(BaseModel):
"""Timeline event for parent dashboard"""
id: str
type: str
text: str
time_ago: str
timestamp: datetime
class ParentRestriction(BaseModel):
"""Restriction summary for parent dashboard"""
event_id: str
text: str
added_at: str
timestamp: datetime
class ParentCaseSummary(BaseModel):
"""Case summary for parent dashboard"""
id: str
injury: str
severity: str
attention: str
status: str
phase_label: str
total_phases: int
opened_at: datetime
milestones: List[ParentMilestoneSummary] = []
timeline: List[ParentTimelineItem] = []
class ParentNextAppointment(BaseModel):
"""Next appointment for parent dashboard"""
date: str
title: str
provider: str
class ParentCareTeamMember(BaseModel):
"""Care team member summary"""
name: str
role: str
class ParentRecentActivity(BaseModel):
"""Recent activity item for parent dashboard"""
type: str
text: str
time_ago: str
class ParentAthleteInfo(BaseModel):
"""Athlete info for parent dashboard"""
id: str
first_name: str
last_name: str
sport: str
team: Optional[str] = None
grade: int
initials: str
class ParentDashboardResponse(BaseModel):
"""Parent dashboard response for a specific child"""
athlete: ParentAthleteInfo
case: Optional[ParentCaseSummary] = None
restrictions: List[ParentRestriction] = []
next_appointment: Optional[ParentNextAppointment] = None
care_team: List[ParentCareTeamMember] = []
recent_activity: List[ParentRecentActivity] = []
============== COLOR HELPERS ==============
AVATAR_COLORS = [
"teal", "blue", "indigo", "purple", "pink",
"rose", "orange", "amber", "emerald", "cyan"
]
def get_avatar_color(athlete_id: str) -> str:
"""Deterministic color based on athlete_id"""
hash_val = sum(ord(c) for c in athlete_id)
return AVATAR_COLORS[hash_val % len(AVATAR_COLORS)]
def get_initials(first_name: str, last_name: str) -> str:
"""Generate initials from first and last name"""
first = first_name[0] if first_name else ""
last = last_name[0] if last_name else ""
return f"{first}{last}".upper()
def truncate_injury(title: str, max_len: int = 30) -> str:
"""Truncate injury title for display"""
if len(title) <= max_len:
return title
return title[:max_len-3] + "..."
def format_time_ago(dt: datetime) -> str:
"""Format datetime as time ago string"""
now = datetime.utcnow()
diff = now - dt
if diff < timedelta(minutes=1):
return "just now"
elif diff < timedelta(hours=1):
minutes = int(diff.seconds / 60)
return f"{minutes}m ago"
elif diff < timedelta(days=1):
hours = int(diff.seconds / 3600)
return f"{hours}h ago"
elif diff < timedelta(days=7):
days = diff.days
return f"{days}d ago"
else:
return dt.strftime("%b %d")
============== ROUTER ENDPOINTS ==============
@router.get("/at", response_model=ATDashboardResponse)
async def get_at_dashboard(
request: Request,
school_id: str = Query(..., description="School identifier for tenant isolation"),
current_user: User = Depends(require_role(["at", "admin"])),
db: Session = Depends(get_db),
):
"""
Get AT dashboard data
Returns:
- stats: Full, modified, and out counts
- active_cases: List of active cases with athlete summaries
- recent_activity: Recent events (clearances, new injuries, etc.)
"""
# Validate school matches user's school
if current_user.school_id != school_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Access denied for this school"
)
# Validate school exists
school = db.query(School).filter(School.id == school_id).first()
if not school:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"School '{school_id}' not found"
)
# Get all athletes for this school
athletes = db.query(Athlete).filter(Athlete.school_id == school_id).all()
athlete_map = {a.id: a for a in athletes}
# Calculate stats
full_count = sum(1 for a in athletes if a.current_status == "cleared")
modified_count = sum(1 for a in athletes if a.current_status == "restricted")
out_count = sum(1 for a in athletes if a.current_status == "out")
stats = DashboardStats(
full_count=full_count,
modified_count=modified_count,
out_count=out_count
)
# Get active cases with athlete info
active_cases_db = db.query(Case).filter(
and_(Case.school_id == school_id, Case.status == "active")
).order_by(Case.opened_at.desc()).all()
active_cases = []
for case in active_cases_db:
athlete = athlete_map.get(case.athlete_id)
if athlete:
# Get current phase info from milestones
milestones = db.query(Milestone).filter(
Milestone.case_id == case.id
).order_by(Milestone.target_date).all()
current_phase = 1
total_phases = len(milestones) if milestones else 4
for i, m in enumerate(milestones, 1):
if m.status == "pending":
current_phase = i
break
elif m.status == "achieved":
current_phase = i + 1
current_phase = min(current_phase, total_phases)
active_cases.append(CaseSummary(
athlete_id=athlete.id,
athlete_name=f"{athlete.first_name} {athlete.last_name}",
initials=get_initials(athlete.first_name, athlete.last_name),
sport=athlete.sports[0] if athlete.sports else "Unknown",
injury=case.title,
injury_short=truncate_injury(case.title),
status=case.status,
status_label=f"Phase {current_phase}/{total_phases}",
avatar_color=get_avatar_color(athlete.id)
))
# Get recent activity (last 7 days)
seven_days_ago = datetime.utcnow() - timedelta(days=7)
recent_events = db.query(Event).filter(
and_(
Event.school_id == school_id,
Event.timestamp >= seven_days_ago
)
).order_by(Event.timestamp.desc()).limit(20).all()
recent_activity = []
for event in recent_events:
athlete = athlete_map.get(event.case.athlete_id) if event.case else None
if athlete:
if event.event_type == "clearance_granted":
recent_activity.append(ActivityItem(
type="cleared",
text=f"{athlete.first_name} {athlete.last_name} cleared for full practice",
time_ago=format_time_ago(event.timestamp)
))
elif event.event_type == "restriction_added":
# Check if it was a removal
if "removed from play" in str(event.content.get("text", "")).lower():
recent_activity.append(ActivityItem(
type="out",
text=f"{athlete.first_name} {athlete.last_name} removed from play",
time_ago=format_time_ago(event.timestamp)
))
else:
recent_activity.append(ActivityItem(
type="modified",
text=f"{athlete.first_name} {athlete.last_name} activity modified",
time_ago=format_time_ago(event.timestamp)
))
elif event.event_type == "note":
recent_activity.append(ActivityItem(
type="modified",
text=f"{athlete.first_name} {athlete.last_name} note added",
time_ago=format_time_ago(event.timestamp)
))
# Limit to 10 items
if len(recent_activity) >= 10:
break
return ATDashboardResponse(
stats=stats,
active_cases=active_cases,
recent_activity=recent_activity
)
============== COACH DASHBOARD ENDPOINT ==============
@router.get("/coach", response_model=CoachDashboardResponse)
async def get_coach_dashboard(
request: Request,
sport: str = Query(..., description="Sport to view (e.g., Football)"),
school_id: str = Query(..., description="School identifier for tenant isolation"),
current_user: User = Depends(require_coach_or_ad),
db: Session = Depends(get_db),
):
"""
Get Coach dashboard data for a specific sport
- Coaches see only their assigned sports
- ADs can see all sports
- Returns sport-scoped roster by team
"""
# Validate school matches user's school
if current_user.school_id != school_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Access denied for this school"
)
# Validate school exists
school = db.query(School).filter(School.id == school_id).first()
if not school:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"School '{school_id}' not found"
)
# Check sport access: coaches can only see assigned sports, ADs can see all
if current_user.role == "coach":
assigned_sports = current_user.assigned_sports or []
if sport not in assigned_sports:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Coach not assigned to sport: {sport}"
)
# Get all athletes for this sport
athletes = db.query(Athlete).filter(
and_(
Athlete.school_id == school_id,
Athlete.sports.contains([sport])
)
).all()
# Get active cases for these athletes
athlete_ids = [a.id for a in athletes]
active_cases = db.query(Case).filter(
and_(
Case.school_id == school_id,
Case.status == "active",
Case.athlete_id.in_(athlete_ids)
)
).all()
# Map athlete_id to case
athlete_case_map = {c.athlete_id: c for c in active_cases}
# Calculate stats
out_count = sum(1 for a in athletes if a.current_status == "out")
modified_count = sum(1 for a in athletes if a.current_status == "restricted")
cleared_count = sum(1 for a in athletes if a.current_status == "cleared")
stats = CoachDashboardStats(
out=out_count,
modified=modified_count,
cleared=cleared_count
)
# Group athletes by team
teams: Dict[str, List[AthleteSummary]] = {"Varsity": [], "JV": [], "Freshman": []}
for athlete in athletes:
team = athlete.team if athlete.team else "Unassigned"
if team not in teams:
teams[team] = []
# Get status label and case info
case = athlete_case_map.get(athlete.id)
status_label = "Cleared"
case_title = None
if athlete.current_status == "out":
status_label = "Out"
case_title = case.title if case else "Unknown"
elif athlete.current_status == "restricted":
status_label = "Modified"
case_title = case.title if case else "Unknown"
teams[team].append(AthleteSummary(
id=athlete.id,
name=f"{athlete.first_name} {athlete.last_name}",
grade=athlete.grade,
position=None, # Could be extended with position data
current_status=athlete.current_status,
status_label=status_label,
case_title=case_title,
avatar_color=get_avatar_color(athlete.id)
))
# Sort athletes in each team by name
for team in teams:
teams[team].sort(key=lambda x: x.name)
# Get recent activity for this sport (last 7 days)
seven_days_ago = datetime.utcnow() - timedelta(days=7)
recent_events = db.query(Event).filter(
and_(
Event.school_id == school_id,
Event.timestamp >= seven_days_ago
)
).order_by(Event.timestamp.desc()).limit(20).all()
# Filter events to sport athletes only
athlete_map = {a.id: a for a in athletes}
recent_activity = []
for event in recent_events:
if event.case and event.case.athlete_id in athlete_map:
athlete = athlete_map[event.case.athlete_id]
if event.event_type == "clearance_granted":
recent_activity.append(ActivityItem(
type="cleared",
text=f"{athlete.first_name} {athlete.last_name} cleared for full practice",
time_ago=format_time_ago(event.timestamp)
))
elif event.event_type == "restriction_added":
if "removed from play" in str(event.content.get("text", "")).lower():
recent_activity.append(ActivityItem(
type="out",
text=f"{athlete.first_name} {athlete.last_name} removed from play",
time_ago=format_time_ago(event.timestamp)
))
else:
recent_activity.append(ActivityItem(
type="modified",
text=f"{athlete.first_name} {athlete.last_name} activity modified",
time_ago=format_time_ago(event.timestamp)
))
elif event.event_type == "note":
recent_activity.append(ActivityItem(
type="modified",
text=f"{athlete.first_name} {athlete.last_name} note added",
time_ago=format_time_ago(event.timestamp)
))
if len(recent_activity) >= 10:
break
return CoachDashboardResponse(
sport=sport,
stats=stats,
teams=teams,
recent_activity=recent_activity
)
============== PARENT DASHBOARD ENDPOINTS ==============
@router.get("/parent/children", response_model=ParentChildrenResponse)
async def get_parent_children(
request: Request,
school_id: str = Query(..., description="School identifier for tenant isolation"),
current_user: User = Depends(require_role(["parent"])),
db: Session = Depends(get_db),
):
"""
Get list of children for the current parent
Returns athletes where parent_ids array contains current user's ID
"""
# Validate school matches user's school
if current_user.school_id != school_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Access denied for this school"
)
# Validate school exists
school = db.query(School).filter(School.id == school_id).first()
if not school:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"School '{school_id}' not found"
)
# Find athletes where parent_ids contains current user's ID
# SQLite stores ARRAY as JSON string in a TEXT column
all_athletes = db.query(Athlete).filter(
Athlete.school_id == school_id
).all()
# Filter by parent_ids containing current user
user_id = current_user.id
children = []
for athlete in all_athletes:
pids = athlete.parent_ids or []
if user_id in pids:
has_active = athlete.current_status in ("restricted", "out")
children.append(ParentChildSummary(
id=athlete.id,
first_name=athlete.first_name,
last_name=athlete.last_name,
sport=athlete.sports[0] if athlete.sports else "Unknown",
team=athlete.team,
grade=athlete.grade,
has_active_case=has_active,
current_status=athlete.current_status,
initials=get_initials(athlete.first_name, athlete.last_name),
avatar_color=get_avatar_color(athlete.id)
))
# Sort by last_name ascending
children.sort(key=lambda c: (c.last_name.lower(), c.first_name.lower()))
return ParentChildrenResponse(children=children)
@router.get("/parent", response_model=ParentDashboardResponse)
async def get_parent_dashboard(
request: Request,
school_id: str = Query(..., description="School identifier"),
athlete_id: str = Query(..., description="Athlete/child ID to view"),
current_user: User = Depends(require_role(["parent"])),
db: Session = Depends(get_db),
):
"""
Get full parent dashboard for a specific child
Returns athlete info, active case with milestones/timeline,
restrictions, upcoming appointments, care team, recent activity
"""
# Validate school matches user's school
if current_user.school_id != school_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Access denied for this school"
)
# Validate school exists
school = db.query(School).filter(School.id == school_id).first()
if not school:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"School '{school_id}' not found"
)
# Get athlete
athlete = db.query(Athlete).filter(
and_(Athlete.id == athlete_id, Athlete.school_id == school_id)
).first()
if not athlete:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Athlete '{athlete_id}' not found"
)
# Verify parent has access to this athlete
if current_user.id not in (athlete.parent_ids or []):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You can only view your own child's dashboard"
)
athlete_info = ParentAthleteInfo(
id=athlete.id,
first_name=athlete.first_name,
last_name=athlete.last_name,
sport=athlete.sports[0] if athlete.sports else "Unknown",
team=athlete.team,
grade=athlete.grade,
initials=get_initials(athlete.first_name, athlete.last_name)
)
# Get active case
active_case = db.query(Case).filter(
and_(Case.athlete_id == athlete_id, Case.school_id == school_id, Case.status == "active")
).first()
case_data = None
restrictions = []
recent_activity = []
care_team = []
next_appointment = None
if active_case:
# Get milestones
milestones = db.query(Milestone).filter(
Milestone.case_id == active_case.id
).order_by(Milestone.target_date).all()
total_phases = max(len(milestones), 4)
# Determine current phase
current_phase = 1
for i, m in enumerate(milestones, 1):
if m.status == "pending":
current_phase = i
break
elif m.status == "achieved":
current_phase = i + 1
current_phase = min(current_phase, total_phases)
# Get timeline events (visible to parent)
all_events = db.query(Event).filter(
Event.case_id == active_case.id
).order_by(Event.timestamp.desc()).all()
# Filter to parent-visible events
parent_events = [e for e in all_events if "parent" in (e.visibility or [])]
timeline_items = []
for e in parent_events:
text = e.content.get("text", "") if e.content else ""
timeline_items.append(ParentTimelineItem(
id=e.id,
type=e.event_type,
text=text,
time_ago=format_time_ago(e.timestamp),
timestamp=e.timestamp
))
# Milestone summaries
milestone_items = []
for m in milestones:
milestone_items.append(ParentMilestoneSummary(
id=m.id,
title=m.title,
status=m.status,
target_date=m.target_date,
achieved_at=m.achieved_at
))
# Restrictions (restriction_added events)
restriction_events = [e for e in all_events if e.event_type == "restriction_added"]
for e in restriction_events:
text = e.content.get("text", "") if e.content else ""
if not text:
text = "Activity modified"
restrictions.append(ParentRestriction(
event_id=e.id,
text=text,
added_at=format_time_ago(e.timestamp),
timestamp=e.timestamp
))
# Care team: the primary AT
if active_case.primary_at_id:
at_user = db.query(User).filter(User.id == active_case.primary_at_id).first()
if at_user:
care_team.append(ParentCareTeamMember(
name=f"{at_user.first_name} {at_user.last_name}",
role="Athletic Trainer"
))
# Recent activity (last 5 parent-visible events)
recent_events = parent_events[:5]
for e in recent_events:
text = e.content.get("text", "") if e.content else ""
recent_activity.append(ParentRecentActivity(
type=e.event_type,
text=text,
time_ago=format_time_ago(e.timestamp)
))
case_data = ParentCaseSummary(
id=active_case.id,
injury=active_case.title,
severity=active_case.severity,
attention=active_case.attention_level,
status=active_case.status,
phase_label=f"Phase {current_phase}/{total_phases}",
total_phases=total_phases,
opened_at=active_case.opened_at,
milestones=milestone_items,
timeline=timeline_items
)
return ParentDashboardResponse(
athlete=athlete_info,
case=case_data,
restrictions=restrictions,
next_appointment=next_appointment,
care_team=care_team,
recent_activity=recent_activity
)
============== AD DASHBOARD ENDPOINT ==============
@router.get("/ad", response_model=ADDashboardResponse)
async def get_ad_dashboard(
request: Request,
school_id: str = Query(..., description="School identifier for tenant isolation"),
current_user: User = Depends(require_ad),
db: Session = Depends(get_db),
):
"""
Get AD (Athletic Director) dashboard overview
- Multi-sport summary view
- All sports in school
- Cross-sport recent activity
"""
# Validate school matches user's school
if current_user.school_id != school_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Access denied for this school"
)
# Validate school exists
school = db.query(School).filter(School.id == school_id).first()
if not school:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"School '{school_id}' not found"
)
# Get all athletes for school
athletes = db.query(Athlete).filter(Athlete.school_id == school_id).all()
# Calculate totals
total_athletes = len(athletes)
out_count = sum(1 for a in athletes if a.current_status == "out")
modified_count = sum(1 for a in athletes if a.current_status == "restricted")
active_cases = db.query(Case).filter(
and_(Case.school_id == school_id, Case.status == "active")
).count()
# Build sport summaries
# Get unique sports from all athletes
all_sports = set()
for athlete in athletes:
for s in athlete.sports:
all_sports.add(s)
sport_summaries = []
for sport in sorted(all_sports):
sport_athletes = [a for a in athletes if sport in a.sports]
sport_out = sum(1 for a in sport_athletes if a.current_status == "out")
sport_modified = sum(1 for a in sport_athletes if a.current_status == "restricted")
sport_summaries.append(SportSummary(
name=sport,
athletes=len(sport_athletes),
out=sport_out,
modified=sport_modified
))
# Sort by athlete count descending
sport_summaries.sort(key=lambda x: x.athletes, reverse=True)
# Get recent activity (last 7 days) - cross-sport
seven_days_ago = datetime.utcnow() - timedelta(days=7)
recent_events = db.query(Event).filter(
and_(
Event.school_id == school_id,
Event.timestamp >= seven_days_ago
)
).order_by(Event.timestamp.desc()).limit(20).all()
athlete_map = {a.id: a for a in athletes}
recent_activity = []
for event in recent_events:
if event.case and event.case.athlete_id in athlete_map:
athlete = athlete_map[event.case.athlete_id]
sport = athlete.sports[0] if athlete.sports else "Unknown"
if event.event_type == "clearance_granted":
recent_activity.append(ActivityItem(
type="cleared",
text=f"{athlete.first_name} {athlete.last_name} ({sport}) cleared for full practice",
time_ago=format_time_ago(event.timestamp)
))
elif event.event_type == "restriction_added":
if "removed from play" in str(event.content.get("text", "")).lower():
recent_activity.append(ActivityItem(
type="out",
text=f"{athlete.first_name} {athlete.last_name} ({sport}) removed from play",
time_ago=format_time_ago(event.timestamp)
))
else:
recent_activity.append(ActivityItem(
type="modified",
text=f"{athlete.first_name} {athlete.last_name} ({sport}) activity modified",
time_ago=format_time_ago(event.timestamp)
))
elif event.event_type == "note":
recent_activity.append(ActivityItem(
type="modified",
text=f"{athlete.first_name} {athlete.last_name} ({sport}) note added",
time_ago=format_time_ago(event.timestamp)
))
if len(recent_activity) >= 10:
break
return ADDashboardResponse(
school_id=school_id,
total_athletes=total_athletes,
active_cases=active_cases,
out_today=out_count,
modified=modified_count,
sports=sport_summaries,
recent_activity=recent_activity
)
============== QUICK STATUS UPDATE SCHEMAS ==============
class QuickStatusUpdateRequest(BaseModel):
"""Quick status update request from AT roster"""
status: str = Field(..., description="New status: cleared|restricted|out|modified")
note: Optional[str] = Field(default=None, description="Optional note for the event")
class QuickStatusUpdateResponse(BaseModel):
"""Response for quick status update"""
athlete: AthleteSchema
event: EventSchema
============== QUICK STATUS UPDATE ENDPOINT ==============
VALID_STATUS_TRANSITIONS = {"cleared", "restricted", "out", "modified"}
@router.patch("/athlete/{athlete_id}/status", response_model=QuickStatusUpdateResponse)
async def quick_status_update(
request: Request,
athlete_id: str,
status_data: QuickStatusUpdateRequest,
school_id: str = Query(..., description="School identifier for tenant isolation"),
current_user: User = Depends(require_role(["at"])),
db: Session = Depends(get_db),
):
"""
Quick Status Update - single-tap action from AT roster
Updates an athlete's current_status and creates a status_change Event.
Status transitions:
- cleared: Full participation
- restricted: Activity modified (limited participation)
- out: Removed from play / not participating
- modified: Activity modification (alias for restricted)
"""
# Validate school
if current_user.school_id != school_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Access denied for this school"
)
# Validate status value
if status_data.status not in VALID_STATUS_TRANSITIONS:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid status: '{status_data.status}'. Must be one of: cleared, restricted, out, modified"
)
# Normalize 'modified' -> 'restricted' for storage
normalized_status = "restricted" if status_data.status == "modified" else status_data.status
# Get athlete
athlete = db.query(Athlete).filter(
and_(Athlete.id == athlete_id, Athlete.school_id == school_id)
).first()
if not athlete:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Athlete '{athlete_id}' not found in school '{school_id}'"
)
# Get active case (or most recent case) for the athlete
active_case = db.query(Case).filter(
and_(
Case.athlete_id == athlete_id,
Case.school_id == school_id,
Case.status == "active"
)
).order_by(Case.opened_at.desc()).first()
# If no active case and we're setting restricted/out, create minimal case
if not active_case and normalized_status in ("restricted", "out"):
case_title = "Status update"
if status_data.note:
case_title = status_data.note
active_case = Case(
id=generate_id("cas_"),
school_id=school_id,
athlete_id=athlete_id,
title=case_title,
severity="mild",
attention_level="stable",
status="active",
opened_at=datetime.utcnow(),
primary_at_id=current_user.id,
created_at=datetime.utcnow(),
updated_at=datetime.utcnow(),
)
db.add(active_case)
db.commit()
db.refresh(active_case)
# Update athlete's active_case_ids
if active_case.id not in (athlete.active_case_ids or []):
athlete.active_case_ids = (athlete.active_case_ids or []) + [active_case.id]
# If setting cleared and there's an active case, close it
if normalized_status == "cleared" and active_case:
active_case.status = "resolved"
active_case.resolved_at = datetime.utcnow()
active_case.updated_at = datetime.utcnow()
# Capture previous status before updating
previous_status = athlete.current_status
# Update athlete status
athlete.current_status = normalized_status
athlete.updated_at = datetime.utcnow()
db.commit()
# Create event
text = status_data.note or f"Status updated to {status_data.status}"
event = Event(
id=generate_id("evt_"),
school_id=school_id,
case_id=active_case.id if active_case else None,
author_id=current_user.id,
event_type="status_change",
visibility=["at", "coach"],
content={
"text": text,
"previous_status": previous_status,
"new_status": normalized_status,
},
timestamp=datetime.utcnow(),
created_at=datetime.utcnow(),
)
db.add(event)
db.commit()
db.refresh(event)
return QuickStatusUpdateResponse(
athlete=AthleteSchema.model_validate(athlete),
event=EventSchema.model_validate(event)
)
============== GENERAL EVENT (JSON body) ==============
class GeneralEventRequest(BaseModel):
"""General event creation with JSON body"""
case_id: str
event_type: str = "note"
note: str
visibility: List[str] = ["at", "coach"]
@router.post("/events/general", status_code=status.HTTP_201_CREATED)
async def create_general_event(
request: Request,
event_data: GeneralEventRequest,
school_id: str = Query(..., description="School identifier"),
current_user: User = Depends(require_role(["at"])),
db: Session = Depends(get_db),
):
"""
Create a general event (note) on a case — JSON body version.
This is called by the AT dashboard frontend's logCaseNote().
The events router has a query-param version; this accepts a JSON body.
"""
if current_user.school_id != school_id:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied")
case = db.query(Case).filter(
and_(Case.id == event_data.case_id, Case.school_id == school_id)
).first()
if not case:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Case not found")
valid_types = {"note", "status_change", "restriction_added", "clearance_granted"}
if event_data.event_type not in valid_types:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid event_type: '{event_data.event_type}'")
for v in event_data.visibility:
if v not in {"at", "coach", "parent"}:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid visibility role: '{v}'")
event = Event(
id=generate_id("evt_"),
school_id=school_id,
case_id=event_data.case_id,
author_id=current_user.id,
event_type=event_data.event_type,
visibility=event_data.visibility,
content={"text": event_data.note},
timestamp=datetime.utcnow(),
created_at=datetime.utcnow(),
)
db.add(event)
db.commit()
db.refresh(event)
return {
"id": event.id,
"case_id": event.case_id,
"event_type": event.event_type,
"note": event_data.note,
"visibility": event_data.visibility,
"timestamp": event.timestamp.isoformat()
}