"""
SQLAlchemy Models for RTSport Database
Maps to Pydantic schemas in schemas.py
PostgreSQL types (JSONB, ARRAY) with SQLite fallback for local testing.
"""
from datetime import datetime
from typing import Optional, List
import json
from sqlalchemy import (
Column, String, Integer, DateTime, Boolean, Text, ForeignKey, Index, event, TypeDecorator, Text as SQLText
)
from sqlalchemy.orm import declarative_base, relationship
import os
Database-aware type selection
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./rtsport_test.db")
IS_SQLITE = DATABASE_URL.startswith("sqlite")
if IS_SQLITE:
# SQLite-compatible types
from sqlalchemy import JSON
JSONB = JSON
# Custom ARRAY type for SQLite (stores as JSON string)
class ARRAY(TypeDecorator):
impl = SQLText
def __init__(self, item_type, **kwargs):
self.item_type = item_type
super().__init__(**kwargs)
def process_bind_param(self, value, dialect):
if value is None:
return None
return json.dumps(list(value) if value else [])
def process_result_value(self, value, dialect):
if value is None:
return []
try:
result = json.loads(value)
return result if isinstance(result, list) else []
except (json.JSONDecodeError, TypeError):
return []
else:
# PostgreSQL native types
from sqlalchemy.dialects.postgresql import JSONB, ARRAY
Base = declarative_base()
class User(Base):
"""User authentication model with role-based access control"""
tablename = "users"
id = Column(String, primary_key=True)
school_id = Column(String, ForeignKey("schools.id", ondelete="CASCADE"), nullable=False)
email = Column(String(255), unique=True, nullable=False)
hashed_password = Column(String(255), nullable=False)
role = Column(String(20), nullable=False) # "at" | "coach" | "parent" | "admin"
first_name = Column(String(100))
last_name = Column(String(100))
is_active = Column(Boolean, default=True)
assigned_sports = Column(ARRAY(String), default=list) # Coach's assigned sports ["Football", "Basketball"]
assigned_teams = Column(ARRAY(String), default=list) # Coach's assigned teams ["Varsity", "JV"]
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
school = relationship("School", back_populates="users")
__table_args__ = (
Index("ix_users_school_id", "school_id"),
Index("ix_users_email", "email"),
Index("ix_users_role", "role"),
Index("ix_users_assigned_sports", "assigned_sports", postgresql_using="gin"),
)
class School(Base):
"""Multi-tenant boundary - every model references school_id"""
tablename = "schools"
id = Column(String, primary_key=True)
name = Column(String(255), nullable=False)
domain = Column(String(255), nullable=False, unique=True)
config = Column(JSONB, default=dict)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
users = relationship("User", back_populates="school", cascade="all, delete-orphan")
athletes = relationship("Athlete", back_populates="school", cascade="all, delete-orphan")
cases = relationship("Case", back_populates="school", cascade="all, delete-orphan")
milestones = relationship("Milestone", back_populates="school", cascade="all, delete-orphan")
events = relationship("Event", back_populates="school", cascade="all, delete-orphan")
__table_args__ = (
Index("ix_schools_domain", "domain"),
)
class Athlete(Base):
"""
Central node - roster queries pull this fast
current_status is cached and updated via DB trigger
"""
tablename = "athletes"
id = Column(String, primary_key=True)
school_id = Column(String, ForeignKey("schools.id", ondelete="CASCADE"), nullable=False)
first_name = Column(String(100), nullable=False)
last_name = Column(String(100), nullable=False)
grade = Column(Integer, nullable=False) # 9-12
sports = Column(ARRAY(String), default=list)
team = Column(String(50), nullable=True)
parent_ids = Column(ARRAY(String), default=list)
current_status = Column(String(20), default="cleared") # cleared|restricted|out
active_case_ids = Column(ARRAY(String), default=list)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
school = relationship("School", back_populates="athletes")
cases = relationship("Case", back_populates="athlete", cascade="all, delete-orphan")
__table_args__ = (
Index("ix_athletes_school_id", "school_id"),
Index("ix_athletes_current_status", "current_status"),
Index("ix_athletes_sports", "sports", postgresql_using="gin"),
Index("ix_athletes_team", "team"),
)
class Case(Base):
"""
Single injury lifecycle
- severity: clinical diagnosis (static: mild|moderate|severe)
- attention_level: operational priority (updated daily: urgent|warning|stable)
- status: active|resolved
"""
tablename = "cases"
id = Column(String, primary_key=True)
school_id = Column(String, ForeignKey("schools.id", ondelete="CASCADE"), nullable=False)
athlete_id = Column(String, ForeignKey("athletes.id", ondelete="CASCADE"), nullable=False)
title = Column(String(255), nullable=False)
severity = Column(String(20), nullable=False) # mild|moderate|severe
attention_level = Column(String(20), default="stable") # urgent|warning|stable
status = Column(String(20), default="active") # active|resolved
opened_at = Column(DateTime, default=datetime.utcnow, nullable=False)
# Case reopening support (30-day window)
reopened_from_case_id = Column(String, ForeignKey("cases.id", ondelete="SET NULL"), nullable=True)
reopened_at = Column(DateTime, nullable=True)
resolved_at = Column(DateTime, nullable=True)
primary_at_id = Column(String, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
school = relationship("School", back_populates="cases")
athlete = relationship("Athlete", back_populates="cases")
milestones = relationship("Milestone", back_populates="case", cascade="all, delete-orphan")
events = relationship("Event", back_populates="case", cascade="all, delete-orphan")
# Self-referential relationship for case reopening
original_case = relationship("Case", remote_side="Case.id",
backref="reopened_cases",
foreign_keys="Case.reopened_from_case_id")
__table_args__ = (
Index("ix_cases_school_id", "school_id"),
Index("ix_cases_athlete_id", "athlete_id"),
Index("ix_cases_status", "status"),
Index("ix_cases_attention_level", "attention_level"),
Index("ix_cases_opened_at", "opened_at"),
)
class Milestone(Base):
"""Forward-looking targets for Phase Bar UI"""
tablename = "milestones"
id = Column(String, primary_key=True)
school_id = Column(String, ForeignKey("schools.id", ondelete="CASCADE"), nullable=False)
case_id = Column(String, ForeignKey("cases.id", ondelete="CASCADE"), nullable=False)
title = Column(String(255), nullable=False)
target_date = Column(DateTime, nullable=False)
status = Column(String(20), default="pending") # pending|achieved|skipped
achieved_at = Column(DateTime, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
school = relationship("School", back_populates="milestones")
case = relationship("Case", back_populates="milestones")
__table_args__ = (
Index("ix_milestones_school_id", "school_id"),
Index("ix_milestones_case_id", "case_id"),
Index("ix_milestones_target_date", "target_date"),
Index("ix_milestones_status", "status"),
)
class Event(Base):
"""
Every sideline entry, clinical note, or clearance
visibility defaults to ["at"] - must explicitly expand
clinical_notes stripped for non-AT roles at API layer
"""
tablename = "events"
id = Column(String, primary_key=True)
school_id = Column(String, ForeignKey("schools.id", ondelete="CASCADE"), nullable=False)
case_id = Column(String, ForeignKey("cases.id", ondelete="CASCADE"), nullable=False)
author_id = Column(String, nullable=False)
event_type = Column(String(50), nullable=False) # note|status_change|restriction_added|clearance_granted
visibility = Column(ARRAY(String), default=list) # ["at"]|["at","coach"]|["at","coach","parent"]
content = Column(JSONB, default=dict) # {text: str, clinical_notes: Optional[str]}
timestamp = Column(DateTime, default=datetime.utcnow, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
# Relationships
school = relationship("School", back_populates="events")
case = relationship("Case", back_populates="events")
__table_args__ = (
Index("ix_events_school_id", "school_id"),
Index("ix_events_case_id", "case_id"),
Index("ix_events_author_id", "author_id"),
Index("ix_events_event_type", "event_type"),
Index("ix_events_timestamp", "timestamp"),
Index("ix_events_visibility", "visibility", postgresql_using="gin"),
)
============== DATABASE TRIGGER FOR ATHLETE STATUS SYNC ==============
def update_athlete_status_on_event(mapper, connection, target: Event):
"""
SQLAlchemy event listener to sync athlete.current_status when events are created
Trigger logic:
- restriction_added -> restricted
- clearance_granted -> cleared (and close case)
- severity with removed_from_play -> out
"""
# Get athlete_id from case first - needed for all event types
result = connection.execute(
Case.__table__.select().where(Case.__table__.c.id == target.case_id)
).fetchone()
if not result:
return # Case not found, can't update athlete
athlete_id = result.athlete_id
if target.event_type == "clearance_granted":
# Close the case
connection.execute(
Case.__table__.update()
.where(Case.__table__.c.id == target.case_id)
.values(status="resolved", resolved_at=datetime.utcnow())
)
# Update athlete status
connection.execute(
Athlete.__table__.update()
.where(Athlete.__table__.c.id == athlete_id)
.values(current_status="cleared")
)
elif target.event_type == "restriction_added":
# Update athlete status to restricted
connection.execute(
Athlete.__table__.update()
.where(Athlete.__table__.c.id == athlete_id)
.values(current_status="restricted")
)
Register the event listener
event.listen(Event, "after_insert", update_athlete_status_on_event)