"""
Athletes API Endpoints (v0.3 Frontend Spec)
GET /api/v1/athletes/{athlete_id} - Returns AthleteSchema per frontend spec
GET /api/v1/athletes/{athlete_id}/case - Returns combined athlete + active case
GET /api/v1/cases/{case_id} - Returns single case by ID
"""
from typing import List, Optional
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, status, Request
from sqlalchemy.orm import Session
from sqlalchemy import and_
from pydantic import BaseModel
from app.database import get_db
from app.models import Athlete, Case, Event, Milestone, School, User
from app.schemas import AthleteSchema, CaseSchema, MilestoneSchema
from app.auth import get_current_user, require_role, require_at_or_coach
router = APIRouter(tags=["athletes"])
============== SCHEMAS ==============
class TimelineEvent(BaseModel):
"""Timeline event for case display"""
date: str
title: str
note: str
by: str
type: str # assessment, restriction, milestone, clearance, note
visibility: List[str] = ["at", "coach"] # FERPA visibility
class MilestoneDisplay(BaseModel):
"""Milestone for case display"""
num: int
label: str
status: str # done, current, pending
completed: Optional[str] # ISO date or null
note: str
class CaseDetail(BaseModel):
"""Extended case schema matching frontend spec"""
case_id: str
athlete_id: str
injury: str
injury_date: str
body_part: str
side: Optional[str] # Left, Right, null
severity: str # mild, moderate, severe
attention: str # stable, urgent, warning
status: str # active, cleared, referred
phase: int
total_phases: int
phase_label: str
projected_return: Optional[str]
last_updated: str
last_updated_by: str
notified: List[str]
milestones: List[MilestoneDisplay]
timeline: List[TimelineEvent]
class FrontendAthlete(BaseModel):
"""Athlete schema matching frontend spec"""
id: str
first_name: str
last_name: str
sport: str
team: str
grade: int
initials: str
active: bool
class CombinedAthleteCaseResponse(BaseModel):
"""Combined athlete + case response"""
athlete: FrontendAthlete
case: Optional[CaseDetail]
============== HELPER FUNCTIONS ==============
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 extract_body_part(title: str) -> str:
"""Extract body part from case title"""
title_lower = title.lower()
parts = ["ankle", "knee", "shoulder", "wrist", "head", "concussion",
"hip", "back", "neck", "elbow", "hand", "foot", "leg", "arm",
"thigh", "calf", "groin", "chest", "ribs", "abdomen", "face"]
for part in parts:
if part in title_lower:
return part.capitalize()
return "Unknown"
def extract_side(title: str) -> Optional[str]:
"""Extract left/right side from case title"""
title_lower = title.lower()
if "left" in title_lower or " - left" in title_lower:
return "Left"
elif "right" in title_lower or " - right" in title_lower:
return "Right"
return None
def get_phase_label(phase: int, total: int) -> str:
"""Get human-readable phase label"""
labels = {
1: "Initial Assessment",
2: "Active Rehab",
3: "Return to Play",
4: "Full Clearance"
}
return labels.get(phase, f"Phase {phase}")
============== ROUTER ENDPOINTS ==============
@router.get("/{athlete_id}", response_model=FrontendAthlete)
async def get_athlete_frontend(
request: Request,
athlete_id: str,
current_user: User = Depends(require_at_or_coach),
db: Session = Depends(get_db),
):
"""
Get athlete profile in frontend format
Matches the schema expected by the AT dashboard frontend.
"""
# Validate school matches user's school
if current_user.school_id != current_user.school_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Access denied for this school"
)
school_id = current_user.school_id
# Query 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"
)
# Coach: check team access
if current_user.role == "coach":
coach_teams = request.headers.get("X-Coach-Teams", "").split(",")
coach_teams = [t.strip() for t in coach_teams if t.strip()]
if coach_teams and athlete.team not in coach_teams:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Coach does not have access to this athlete's team"
)
return FrontendAthlete(
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 or "",
grade=athlete.grade,
initials=get_initials(athlete.first_name, athlete.last_name),
active=athlete.current_status != "out"
)
@router.get("/{athlete_id}/case", response_model=CombinedAthleteCaseResponse)
async def get_athlete_with_case(
request: Request,
athlete_id: str,
current_user: User = Depends(require_at_or_coach),
db: Session = Depends(get_db),
):
"""
Get combined athlete + active case data
Returns athlete profile and their active case (if any) in a single response.
This saves a round-trip from the frontend which previously made two separate calls.
"""
school_id = current_user.school_id
# Query 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"
)
# Coach: check team access
if current_user.role == "coach":
coach_teams = request.headers.get("X-Coach-Teams", "").split(",")
coach_teams = [t.strip() for t in coach_teams if t.strip()]
if coach_teams and athlete.team not in coach_teams:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Coach does not have access to this athlete's team"
)
# Get active case (most recent)
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()
athlete_data = FrontendAthlete(
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 or "",
grade=athlete.grade,
initials=get_initials(athlete.first_name, athlete.last_name),
active=athlete.current_status != "out"
)
if not active_case:
return CombinedAthleteCaseResponse(athlete=athlete_data, case=None)
# Get milestones for this case
milestones_db = db.query(Milestone).filter(
Milestone.case_id == active_case.id
).order_by(Milestone.target_date).all()
milestones = []
current_phase = 1
for i, m in enumerate(milestones_db, 1):
status = "done" if m.status == "achieved" else "current" if m.status == "pending" and current_phase == i else "pending"
if status == "current":
current_phase = i
milestones.append(MilestoneDisplay(
num=i,
label=m.title,
status=status,
completed=m.achieved_at.strftime("%Y-%m-%d") if m.achieved_at else None,
note=m.title
))
if not milestones:
# Default milestones if none exist
milestones = [
MilestoneDisplay(num=1, label="Initial Assessment", status="done", completed=None, note=""),
MilestoneDisplay(num=2, label="Active Rehab", status="current", completed=None, note=""),
MilestoneDisplay(num=3, label="Return to Play", status="pending", completed=None, note=""),
MilestoneDisplay(num=4, label="Full Clearance", status="pending", completed=None, note=""),
]
current_phase = 2
total_phases = len(milestones)
# Get timeline events
events_db = db.query(Event).filter(
Event.case_id == active_case.id
).order_by(Event.timestamp.desc()).all()
timeline = []
for event in events_db:
event_type_map = {
"note": "note",
"status_change": "milestone",
"restriction_added": "restriction",
"clearance_granted": "clearance"
}
content_text = event.content.get("text", "") if isinstance(event.content, dict) else ""
timeline.append(TimelineEvent(
date=event.timestamp.strftime("%b %d · %I:%M %p"),
title=event.event_type.replace("_", " ").title(),
note=content_text[:100] + "..." if len(content_text) > 100 else content_text,
by=event.author_id[:6] if event.author_id else "You",
type=event_type_map.get(event.event_type, "note"),
visibility=event.visibility or ["at", "coach"]
))
# Build notified list (simplified - would come from event visibility)
notified = ["Coach"]
if active_case.attention_level == "urgent":
notified.append("Parent")
case_detail = CaseDetail(
case_id=active_case.id,
athlete_id=athlete.id,
injury=active_case.title,
injury_date=active_case.opened_at.strftime("%Y-%m-%d"),
body_part=extract_body_part(active_case.title),
side=extract_side(active_case.title),
severity=active_case.severity,
attention=active_case.attention_level,
status=active_case.status,
phase=current_phase,
total_phases=total_phases,
phase_label=get_phase_label(current_phase, total_phases),
projected_return=None, # Would be calculated from milestones
last_updated=active_case.updated_at.strftime("%Y-%m-%d %H:%M"),
last_updated_by="You",
notified=notified,
milestones=milestones,
timeline=timeline
)
return CombinedAthleteCaseResponse(athlete=athlete_data, case=case_detail)