""" 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) # Note: Use app.utils.id_generator.generate_id() for ID generation