"""Initial migration — all MVP tables. Revision ID: 0001_initial Revises: Create Date: 2026-08-10 00:00:00 """ from typing import Sequence, Union from alembic import op import sqlalchemy as sa from sqlalchemy.dialects import postgresql revision: str = "0001_initial" down_revision: Union[str, None] = None branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None def upgrade() -> None: # Roles op.create_table( "roles", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("name", sa.String(50), unique=True, nullable=False), sa.Column("description", sa.Text), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Teams op.create_table( "teams", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("name", sa.String(255), unique=True, nullable=False), sa.Column("description", sa.Text), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Users op.create_table( "users", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("entra_oid", sa.String(128), unique=True, nullable=False), sa.Column("email", sa.String(255), unique=True, nullable=False), sa.Column("full_name", sa.String(255), nullable=False), sa.Column("display_name", sa.String(100)), sa.Column("is_active", sa.Boolean, server_default=sa.true(), nullable=False), sa.Column("team_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("teams.id"), nullable=True), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # User-Roles association op.create_table( "user_roles", sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), primary_key=True), sa.Column("role_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("roles.id"), primary_key=True), ) # Programs op.create_table( "programs", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("name", sa.String(255), nullable=False), sa.Column("description", sa.Text), sa.Column("is_active", sa.Boolean, server_default=sa.true(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Cohorts op.create_table( "cohorts", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("name", sa.String(255), nullable=False), sa.Column("program_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("programs.id"), nullable=False), sa.Column("batch_size", sa.Integer, server_default="5"), sa.Column("is_active", sa.Boolean, server_default=sa.true(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Cohort members op.create_table( "cohort_members", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=False), sa.Column("cohort_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("cohorts.id"), nullable=False), sa.Column("joined_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.UniqueConstraint("user_id", "cohort_id", name="uq_user_cohort"), ) # Modules op.create_table( "modules", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("program_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("programs.id"), nullable=False), sa.Column("name", sa.String(255), nullable=False), sa.Column("description", sa.Text), sa.Column("order_index", sa.Integer, server_default="0", nullable=False), sa.Column("is_active", sa.Boolean, server_default=sa.true(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Module rules op.create_table( "module_rules", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("module_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("modules.id"), unique=True, nullable=False), sa.Column("pass_threshold", sa.Float, server_default="70.0", nullable=False), sa.Column("is_mandatory", sa.Boolean, server_default=sa.true(), nullable=False), sa.Column("max_replays", sa.Integer), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Source documents op.create_table( "source_documents", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("filename", sa.String(500), nullable=False), sa.Column("file_path", sa.String(1000), nullable=False), sa.Column("file_type", sa.String(20), nullable=False), sa.Column("file_size", sa.Integer, nullable=False), sa.Column("uploaded_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=False), sa.Column("extracted_text", sa.Text), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Quizzes op.create_table( "quizzes", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("module_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("modules.id"), nullable=False), sa.Column("title", sa.String(500), nullable=False), sa.Column("description", sa.Text), sa.Column("status", sa.Enum("draft", "in_review", "published", "archived", name="quizstatus"), server_default="draft", nullable=False), sa.Column("created_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=False), sa.Column("requires_credit_head", sa.Boolean, server_default=sa.false(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Quiz versions op.create_table( "quiz_versions", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("quiz_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("quizzes.id"), nullable=False), sa.Column("version_number", sa.Integer, nullable=False), sa.Column("snapshot", postgresql.JSON, nullable=False), sa.Column("published_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.UniqueConstraint("quiz_id", "version_number", name="uq_quiz_version"), ) # Questions op.create_table( "questions", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("quiz_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("quizzes.id"), nullable=False), sa.Column("question_text", sa.Text, nullable=False), sa.Column("topic_tag", sa.String(255), server_default="general", nullable=False), sa.Column("explanation", sa.Text), sa.Column("confidence_score", sa.Float, server_default="0"), sa.Column("is_ai_generated", sa.Boolean, server_default=sa.false(), nullable=False), sa.Column("source_document_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("source_documents.id"), nullable=True), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Question options op.create_table( "question_options", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("question_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("questions.id"), nullable=False), sa.Column("option_text", sa.Text, nullable=False), sa.Column("is_correct", sa.Boolean, nullable=False, server_default=sa.false()), sa.Column("order_index", sa.Integer, server_default="0", nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Document jobs op.create_table( "document_jobs", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("document_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("source_documents.id"), nullable=False), sa.Column("status", sa.Enum("uploaded", "parsing", "generating", "review_ready", "failed", name="docjobstatus"), server_default="uploaded", nullable=False), sa.Column("error_message", sa.Text), sa.Column("generated_quiz_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("quizzes.id"), nullable=True), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Assignments op.create_table( "assignments", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("quiz_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("quizzes.id"), nullable=False), sa.Column("cohort_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("cohorts.id"), nullable=True), sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=True), sa.Column("due_date", sa.DateTime(timezone=True)), sa.Column("is_active", sa.Boolean, server_default=sa.true(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Attempts op.create_table( "attempts", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("assignment_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("assignments.id"), nullable=False), sa.Column("learner_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=False), sa.Column("module_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("modules.id"), nullable=False), sa.Column("status", sa.Enum("in_progress", "completed", "abandoned", name="attemptstatus"), server_default="in_progress", nullable=False), sa.Column("score", sa.Float, server_default="0"), sa.Column("correct_count", sa.Integer, server_default="0"), sa.Column("total_questions", sa.Integer, server_default="0"), sa.Column("started_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("completed_at", sa.DateTime(timezone=True)), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Attempt answers op.create_table( "attempt_answers", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("attempt_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("attempts.id"), nullable=False), sa.Column("question_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("questions.id"), nullable=False), sa.Column("selected_option_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("question_options.id"), nullable=True), sa.Column("is_correct", sa.Boolean, server_default=sa.false(), nullable=False), sa.Column("answered_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Learner progress op.create_table( "learner_progress", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("learner_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=False), sa.Column("module_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("modules.id"), nullable=False), sa.Column("best_score", sa.Float, server_default="0"), sa.Column("latest_score", sa.Float, server_default="0"), sa.Column("attempt_count", sa.Integer, server_default="0"), sa.Column("is_passed", sa.Boolean, server_default=sa.false(), nullable=False), sa.Column("signed_off", sa.Boolean, server_default=sa.false(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.UniqueConstraint("learner_id", "module_id", name="uq_learner_module"), ) # Weak topic snapshots op.create_table( "weak_topic_snapshots", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("learner_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=False), sa.Column("module_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("modules.id"), nullable=False), sa.Column("topic_tag", sa.String(255), nullable=False), sa.Column("total_questions", sa.Integer, server_default="0"), sa.Column("correct_questions", sa.Integer, server_default="0"), sa.Column("weakness_ratio", sa.Float, server_default="0"), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Manager signoffs op.create_table( "manager_signoffs", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("learner_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=False), sa.Column("module_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("modules.id"), nullable=False), sa.Column("manager_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=False), sa.Column("status", sa.Enum("pending", "approved", "rejected", name="signoffstatus"), server_default="pending", nullable=False), sa.Column("comments", sa.Text), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.UniqueConstraint("learner_id", "module_id", name="uq_signoff_learner_module"), ) # Certificates op.create_table( "certificates", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("learner_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=False), sa.Column("module_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("modules.id"), nullable=False), sa.Column("certificate_number", sa.String(100), unique=True, nullable=False), sa.Column("issued_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Badges op.create_table( "badges", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("learner_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=False), sa.Column("badge_type", sa.String(100), nullable=False), sa.Column("metadata", postgresql.JSON, server_default="{}"), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Leaderboard snapshots op.create_table( "leaderboard_snapshots", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=False), sa.Column("week_key", sa.String(20), nullable=False), sa.Column("rank", sa.Integer, nullable=False), sa.Column("score", sa.Float, server_default="0", nullable=False), sa.Column("streak_count", sa.Integer, server_default="0"), sa.Column("mastery_points", sa.Float, server_default="0"), sa.Column("improvement_points", sa.Float, server_default="0"), sa.Column("consistency_points", sa.Float, server_default="0"), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.UniqueConstraint("user_id", "week_key", name="uq_user_week"), ) op.create_index("ix_leaderboard_week", "leaderboard_snapshots", ["week_key"]) # Audit logs op.create_table( "audit_logs", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), sa.Column("actor_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=True), sa.Column("action", sa.String(100), nullable=False), sa.Column("entity_type", sa.String(100), nullable=False), sa.Column("entity_id", postgresql.UUID(as_uuid=True), nullable=True), sa.Column("details", postgresql.JSON, server_default="{}"), sa.Column("ip_address", sa.String(45)), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), ) # Seed default roles op.execute(""" INSERT INTO roles (id, name, description) VALUES (gen_random_uuid(), 'learner', 'Learner role'), (gen_random_uuid(), 'sales_manager', 'Sales Manager role'), (gen_random_uuid(), 'dept_head', 'Department Head / Trainer role'), (gen_random_uuid(), 'ld_admin', 'L&D Admin role'), (gen_random_uuid(), 'credit_head', 'Credit Head reviewer role'); """) def downgrade() -> None: for table in [ "audit_logs", "leaderboard_snapshots", "badges", "certificates", "manager_signoffs", "weak_topic_snapshots", "learner_progress", "attempt_answers", "attempts", "assignments", "document_jobs", "question_options", "questions", "quiz_versions", "quizzes", "source_documents", "module_rules", "modules", "cohort_members", "cohorts", "user_roles", "users", "teams", "programs", "roles", ]: op.drop_table(table) for enum in ["quizstatus", "docjobstatus", "attemptstatus", "signoffstatus"]: op.execute(f"DROP TYPE IF EXISTS {enum}")