SafePath Training — Phase 1A: Database Schema & Migrations
Overview
This document covers all database infrastructure for SafePath Training:
- 12 new
safepath_tables - Complete Alembic migration SQL
- SQLAlchemy ORM models
- Seed data for course categories and certification types
Prerequisite: None (this is the first phase).
Files to create/modify:
tellus-ehs-hazcom-service/alembic/versions/XXXX_add_safepath_training_tables.py(migration)tellus-ehs-hazcom-service/app/db/models/safepath.py(all models)tellus-ehs-hazcom-service/app/db/models/__init__.py(register models)
Database Tables
Entity Relationship Diagram
safepath_course_categories
│
▼
safepath_courses ◄──────────────── safepath_lessons
│ │
│ ▼
│ safepath_lesson_assets
│
├──── safepath_quizzes
│ │
│ ▼
│ safepath_quiz_questions
│
├──── safepath_assignments ──────► safepath_results
│ │
│ └──── (assigned_to → users)
│
├──── safepath_classroom_sessions
│
└──── safepath_auto_assignment_rules
safepath_certification_types
│
▼
safepath_certifications ◄──── safepath_results (optional link)
safepath_audit_log (standalone, references any entity)
Alembic Migration
File: tellus-ehs-hazcom-service/alembic/versions/XXXX_add_safepath_training_tables.py
Generate with:
cd tellus-ehs-hazcom-service
alembic revision --autogenerate -m "add safepath training tables"
Then replace the generated content with the following reviewed migration:
"""add safepath training tables
Revision ID: <auto-generated>
Revises: <previous-head>
Create Date: <auto-generated>
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers
revision = '<auto-generated>'
down_revision = '<previous-head>'
branch_labels = None
depends_on = None
def upgrade() -> None:
# ================================================================
# 1. safepath_course_categories
# ================================================================
op.create_table(
'safepath_course_categories',
sa.Column('category_id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), nullable=False),
sa.Column('name', sa.String(100), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('is_system', sa.Boolean(), server_default=sa.text('true'), nullable=False),
sa.Column('company_id', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('sort_order', sa.Integer(), server_default=sa.text('0'), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.ForeignKeyConstraint(['company_id'], ['core_data_companies.company_id'], ondelete='CASCADE'),
sa.PrimaryKeyConstraint('category_id'),
)
op.create_index('ix_safepath_course_categories_company', 'safepath_course_categories', ['company_id'])
# ================================================================
# 2. safepath_certification_types
# ================================================================
op.create_table(
'safepath_certification_types',
sa.Column('type_id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), nullable=False),
sa.Column('name', sa.String(150), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('default_validity_months', sa.Integer(), nullable=True),
sa.Column('retraining_frequency_months', sa.Integer(), nullable=True),
sa.Column('osha_standard_ref', sa.String(50), nullable=True),
sa.Column('is_system', sa.Boolean(), server_default=sa.text('true'), nullable=False),
sa.Column('company_id', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.ForeignKeyConstraint(['company_id'], ['core_data_companies.company_id'], ondelete='CASCADE'),
sa.PrimaryKeyConstraint('type_id'),
)
op.create_index('ix_safepath_cert_types_company', 'safepath_certification_types', ['company_id'])
# ================================================================
# 3. safepath_courses
# ================================================================
op.create_table(
'safepath_courses',
sa.Column('course_id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), nullable=False),
sa.Column('company_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('title', sa.String(255), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('category_id', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('osha_standard_ref', sa.String(50), nullable=True),
sa.Column('estimated_duration_minutes', sa.Integer(), nullable=True),
sa.Column('passing_score_percent', sa.Integer(), server_default=sa.text('80'), nullable=False),
sa.Column('max_retakes', sa.Integer(), server_default=sa.text('3'), nullable=False),
sa.Column('status', sa.String(20), server_default=sa.text("'draft'"), nullable=False),
sa.Column('version_number', sa.Integer(), server_default=sa.text('1'), nullable=False),
sa.Column('parent_course_id', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('plan_version_id', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('locale', sa.String(5), server_default=sa.text("'en'"), nullable=False),
sa.Column('created_by', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['company_id'], ['core_data_companies.company_id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['category_id'], ['safepath_course_categories.category_id'], ondelete='SET NULL'),
sa.ForeignKeyConstraint(['parent_course_id'], ['safepath_courses.course_id'], ondelete='SET NULL'),
sa.ForeignKeyConstraint(['created_by'], ['users.user_id'], ondelete='SET NULL'),
sa.CheckConstraint("status IN ('draft', 'published', 'archived')", name='ck_safepath_courses_status'),
sa.CheckConstraint('passing_score_percent >= 0 AND passing_score_percent <= 100', name='ck_safepath_courses_passing_score'),
sa.CheckConstraint('max_retakes >= 0', name='ck_safepath_courses_max_retakes'),
sa.PrimaryKeyConstraint('course_id'),
)
op.create_index('ix_safepath_courses_company', 'safepath_courses', ['company_id'])
op.create_index('ix_safepath_courses_status', 'safepath_courses', ['status'])
op.create_index('ix_safepath_courses_category', 'safepath_courses', ['category_id'])
op.create_index('ix_safepath_courses_company_status', 'safepath_courses', ['company_id', 'status'])
# ================================================================
# 4. safepath_lessons
# ================================================================
op.create_table(
'safepath_lessons',
sa.Column('lesson_id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), nullable=False),
sa.Column('course_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('title', sa.String(255), nullable=False),
sa.Column('lesson_type', sa.String(20), nullable=False),
sa.Column('content', postgresql.JSONB(), nullable=True),
sa.Column('sort_order', sa.Integer(), server_default=sa.text('0'), nullable=False),
sa.Column('completion_threshold_percent', sa.Integer(), server_default=sa.text('80'), nullable=False),
sa.Column('locale', sa.String(5), server_default=sa.text("'en'"), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['course_id'], ['safepath_courses.course_id'], ondelete='CASCADE'),
sa.CheckConstraint("lesson_type IN ('video', 'pdf', 'slides', 'text', 'external_link')", name='ck_safepath_lessons_type'),
sa.PrimaryKeyConstraint('lesson_id'),
)
op.create_index('ix_safepath_lessons_course', 'safepath_lessons', ['course_id'])
op.create_index('ix_safepath_lessons_course_order', 'safepath_lessons', ['course_id', 'sort_order'])
# ================================================================
# 5. safepath_lesson_assets
# ================================================================
op.create_table(
'safepath_lesson_assets',
sa.Column('asset_id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), nullable=False),
sa.Column('lesson_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('file_name', sa.String(255), nullable=False),
sa.Column('s3_bucket', sa.String(100), nullable=False),
sa.Column('s3_key', sa.Text(), nullable=False),
sa.Column('file_type', sa.String(50), nullable=False),
sa.Column('file_size_bytes', sa.BigInteger(), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.ForeignKeyConstraint(['lesson_id'], ['safepath_lessons.lesson_id'], ondelete='CASCADE'),
sa.PrimaryKeyConstraint('asset_id'),
)
op.create_index('ix_safepath_lesson_assets_lesson', 'safepath_lesson_assets', ['lesson_id'])
# ================================================================
# 6. safepath_quizzes
# ================================================================
op.create_table(
'safepath_quizzes',
sa.Column('quiz_id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), nullable=False),
sa.Column('course_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('title', sa.String(255), server_default=sa.text("'Course Quiz'"), nullable=False),
sa.Column('sort_order', sa.Integer(), server_default=sa.text('0'), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['course_id'], ['safepath_courses.course_id'], ondelete='CASCADE'),
sa.PrimaryKeyConstraint('quiz_id'),
)
op.create_index('ix_safepath_quizzes_course', 'safepath_quizzes', ['course_id'])
# ================================================================
# 7. safepath_quiz_questions
# ================================================================
op.create_table(
'safepath_quiz_questions',
sa.Column('question_id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), nullable=False),
sa.Column('quiz_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('question_type', sa.String(20), nullable=False),
sa.Column('question_text', sa.Text(), nullable=False),
sa.Column('options', postgresql.JSONB(), nullable=False),
sa.Column('explanation', sa.Text(), nullable=True),
sa.Column('sort_order', sa.Integer(), server_default=sa.text('0'), nullable=False),
sa.Column('locale', sa.String(5), server_default=sa.text("'en'"), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.ForeignKeyConstraint(['quiz_id'], ['safepath_quizzes.quiz_id'], ondelete='CASCADE'),
sa.CheckConstraint("question_type IN ('mcq_single', 'mcq_multi', 'true_false', 'matching')", name='ck_safepath_questions_type'),
sa.PrimaryKeyConstraint('question_id'),
)
op.create_index('ix_safepath_questions_quiz', 'safepath_quiz_questions', ['quiz_id'])
op.create_index('ix_safepath_questions_quiz_order', 'safepath_quiz_questions', ['quiz_id', 'sort_order'])
# ================================================================
# 8. safepath_assignments
# ================================================================
op.create_table(
'safepath_assignments',
sa.Column('assignment_id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), nullable=False),
sa.Column('company_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('course_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('course_version', sa.Integer(), nullable=False),
sa.Column('assigned_to', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('assigned_by', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('site_id', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('due_date', sa.Date(), nullable=False),
sa.Column('priority', sa.String(10), server_default=sa.text("'normal'"), nullable=False),
sa.Column('status', sa.String(20), server_default=sa.text("'pending'"), nullable=False),
sa.Column('notes', sa.Text(), nullable=True),
sa.Column('auto_rule_id', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['company_id'], ['core_data_companies.company_id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['course_id'], ['safepath_courses.course_id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['assigned_to'], ['users.user_id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['assigned_by'], ['users.user_id'], ondelete='SET NULL'),
sa.ForeignKeyConstraint(['site_id'], ['company_sites.site_id'], ondelete='SET NULL'),
sa.CheckConstraint("status IN ('pending', 'in_progress', 'completed', 'overdue', 'expired')", name='ck_safepath_assignments_status'),
sa.CheckConstraint("priority IN ('normal', 'urgent')", name='ck_safepath_assignments_priority'),
sa.PrimaryKeyConstraint('assignment_id'),
)
op.create_index('ix_safepath_assignments_company', 'safepath_assignments', ['company_id'])
op.create_index('ix_safepath_assignments_assigned_to', 'safepath_assignments', ['assigned_to'])
op.create_index('ix_safepath_assignments_course', 'safepath_assignments', ['course_id'])
op.create_index('ix_safepath_assignments_status', 'safepath_assignments', ['status'])
op.create_index('ix_safepath_assignments_due_date', 'safepath_assignments', ['due_date'])
op.create_index('ix_safepath_assignments_company_status', 'safepath_assignments', ['company_id', 'status'])
op.create_index('ix_safepath_assignments_user_status', 'safepath_assignments', ['assigned_to', 'status'])
# ================================================================
# 9. safepath_results
# ================================================================
op.create_table(
'safepath_results',
sa.Column('result_id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), nullable=False),
sa.Column('assignment_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('user_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('course_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('attempt_number', sa.Integer(), nullable=False),
sa.Column('score_percent', sa.Numeric(5, 2), nullable=True),
sa.Column('passed', sa.Boolean(), nullable=True),
sa.Column('started_at', sa.DateTime(timezone=True), nullable=True),
sa.Column('completed_at', sa.DateTime(timezone=True), nullable=True),
sa.Column('duration_seconds', sa.Integer(), nullable=True),
sa.Column('delivery_method', sa.String(20), server_default=sa.text("'online'"), nullable=False),
sa.Column('instructor_name', sa.String(255), nullable=True),
sa.Column('quiz_answers', postgresql.JSONB(), nullable=True),
sa.Column('lesson_progress', postgresql.JSONB(), nullable=True),
sa.Column('acknowledgment_signature', sa.Text(), nullable=True),
sa.Column('acknowledgment_ip_hash', sa.String(64), nullable=True),
sa.Column('acknowledgment_at', sa.DateTime(timezone=True), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.ForeignKeyConstraint(['assignment_id'], ['safepath_assignments.assignment_id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['user_id'], ['users.user_id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['course_id'], ['safepath_courses.course_id'], ondelete='CASCADE'),
sa.CheckConstraint("delivery_method IN ('online', 'in_person', 'external')", name='ck_safepath_results_delivery'),
sa.CheckConstraint('attempt_number >= 1', name='ck_safepath_results_attempt'),
sa.PrimaryKeyConstraint('result_id'),
)
op.create_index('ix_safepath_results_assignment', 'safepath_results', ['assignment_id'])
op.create_index('ix_safepath_results_user', 'safepath_results', ['user_id'])
op.create_index('ix_safepath_results_course', 'safepath_results', ['course_id'])
op.create_index('ix_safepath_results_user_course', 'safepath_results', ['user_id', 'course_id'])
# ================================================================
# 10. safepath_certifications
# ================================================================
op.create_table(
'safepath_certifications',
sa.Column('certification_id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), nullable=False),
sa.Column('company_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('user_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('certification_type_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('issuing_authority', sa.String(255), nullable=True),
sa.Column('certification_number', sa.String(100), nullable=True),
sa.Column('issue_date', sa.Date(), nullable=False),
sa.Column('expiration_date', sa.Date(), nullable=True),
sa.Column('status', sa.String(20), server_default=sa.text("'active'"), nullable=False),
sa.Column('evidence_s3_bucket', sa.String(100), nullable=True),
sa.Column('evidence_s3_key', sa.Text(), nullable=True),
sa.Column('source', sa.String(20), server_default=sa.text("'external'"), nullable=False),
sa.Column('result_id', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('notes', sa.Text(), nullable=True),
sa.Column('created_by', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['company_id'], ['core_data_companies.company_id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['user_id'], ['users.user_id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['certification_type_id'], ['safepath_certification_types.type_id'], ondelete='RESTRICT'),
sa.ForeignKeyConstraint(['result_id'], ['safepath_results.result_id'], ondelete='SET NULL'),
sa.ForeignKeyConstraint(['created_by'], ['users.user_id'], ondelete='SET NULL'),
sa.CheckConstraint("status IN ('active', 'expiring_soon', 'expired', 'revoked')", name='ck_safepath_certs_status'),
sa.CheckConstraint("source IN ('internal', 'external')", name='ck_safepath_certs_source'),
sa.PrimaryKeyConstraint('certification_id'),
)
op.create_index('ix_safepath_certs_company', 'safepath_certifications', ['company_id'])
op.create_index('ix_safepath_certs_user', 'safepath_certifications', ['user_id'])
op.create_index('ix_safepath_certs_type', 'safepath_certifications', ['certification_type_id'])
op.create_index('ix_safepath_certs_expiration', 'safepath_certifications', ['expiration_date'])
op.create_index('ix_safepath_certs_company_status', 'safepath_certifications', ['company_id', 'status'])
op.create_index('ix_safepath_certs_user_status', 'safepath_certifications', ['user_id', 'status'])
# ================================================================
# 11. safepath_classroom_sessions
# ================================================================
op.create_table(
'safepath_classroom_sessions',
sa.Column('session_id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), nullable=False),
sa.Column('company_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('course_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('instructor_name', sa.String(255), nullable=False),
sa.Column('session_date', sa.Date(), nullable=False),
sa.Column('start_time', sa.Time(), nullable=True),
sa.Column('end_time', sa.Time(), nullable=True),
sa.Column('location', sa.String(255), nullable=True),
sa.Column('site_id', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('max_attendees', sa.Integer(), nullable=True),
sa.Column('notes', sa.Text(), nullable=True),
sa.Column('created_by', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.ForeignKeyConstraint(['company_id'], ['core_data_companies.company_id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['course_id'], ['safepath_courses.course_id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['site_id'], ['company_sites.site_id'], ondelete='SET NULL'),
sa.ForeignKeyConstraint(['created_by'], ['users.user_id'], ondelete='SET NULL'),
sa.PrimaryKeyConstraint('session_id'),
)
op.create_index('ix_safepath_classroom_company', 'safepath_classroom_sessions', ['company_id'])
op.create_index('ix_safepath_classroom_course', 'safepath_classroom_sessions', ['course_id'])
op.create_index('ix_safepath_classroom_date', 'safepath_classroom_sessions', ['session_date'])
# ================================================================
# 12. safepath_auto_assignment_rules
# ================================================================
op.create_table(
'safepath_auto_assignment_rules',
sa.Column('rule_id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), nullable=False),
sa.Column('company_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('course_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('trigger_type', sa.String(30), nullable=False),
sa.Column('trigger_config', postgresql.JSONB(), server_default=sa.text("'{}'::jsonb"), nullable=False),
sa.Column('due_date_offset_days', sa.Integer(), server_default=sa.text('30'), nullable=False),
sa.Column('is_active', sa.Boolean(), server_default=sa.text('true'), nullable=False),
sa.Column('created_by', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['company_id'], ['core_data_companies.company_id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['course_id'], ['safepath_courses.course_id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['created_by'], ['users.user_id'], ondelete='SET NULL'),
sa.CheckConstraint("trigger_type IN ('new_hire', 'role_change', 'site_transfer', 'cert_expiring', 'chemical_added')", name='ck_safepath_rules_trigger'),
sa.PrimaryKeyConstraint('rule_id'),
)
op.create_index('ix_safepath_rules_company', 'safepath_auto_assignment_rules', ['company_id'])
op.create_index('ix_safepath_rules_company_active', 'safepath_auto_assignment_rules', ['company_id', 'is_active'])
# ================================================================
# 13. safepath_audit_log
# ================================================================
op.create_table(
'safepath_audit_log',
sa.Column('log_id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), nullable=False),
sa.Column('company_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('event_type', sa.String(50), nullable=False),
sa.Column('entity_type', sa.String(30), nullable=False),
sa.Column('entity_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('user_id', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('details', postgresql.JSONB(), nullable=True),
sa.Column('ip_hash', sa.String(64), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.ForeignKeyConstraint(['company_id'], ['core_data_companies.company_id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['user_id'], ['users.user_id'], ondelete='SET NULL'),
sa.PrimaryKeyConstraint('log_id'),
)
op.create_index('ix_safepath_audit_company', 'safepath_audit_log', ['company_id'])
op.create_index('ix_safepath_audit_entity', 'safepath_audit_log', ['entity_type', 'entity_id'])
op.create_index('ix_safepath_audit_created', 'safepath_audit_log', ['created_at'])
op.create_index('ix_safepath_audit_event_type', 'safepath_audit_log', ['event_type'])
# ================================================================
# Seed data: Course categories
# ================================================================
op.execute("""
INSERT INTO safepath_course_categories (name, description, is_system, sort_order) VALUES
('Hazard Communication (HazCom/GHS)', 'OSHA 1910.1200 - Hazard Communication Standard including GHS labeling', true, 1),
('Fall Protection', 'OSHA 1926.501/503 - Fall protection requirements and training', true, 2),
('Lockout/Tagout (LOTO)', 'OSHA 1910.147 - Control of hazardous energy', true, 3),
('Respiratory Protection', 'OSHA 1910.134 - Respiratory protection program requirements', true, 4),
('Confined Space Entry', 'OSHA 1910.146 - Permit-required confined spaces', true, 5),
('Electrical Safety', 'OSHA 1910.331-335 - Electrical safety-related work practices', true, 6),
('Personal Protective Equipment (PPE)', 'OSHA 1910.132 - General PPE requirements', true, 7),
('Bloodborne Pathogens', 'OSHA 1910.1030 - Bloodborne pathogen exposure control', true, 8),
('Fire Safety / Emergency Action', 'OSHA 1910.38/39 - Emergency action and fire prevention plans', true, 9),
('Forklift / Powered Industrial Trucks', 'OSHA 1910.178 - Powered industrial truck operator training', true, 10),
('Excavation / Trenching', 'OSHA 1926.651 - Excavation safety requirements', true, 11),
('Scaffold Safety', 'OSHA 1926.451 - Scaffold requirements and training', true, 12),
('General Safety Orientation', 'General workplace safety and new-hire orientation', true, 13),
('Machine Guarding', 'OSHA 1910.212 - General requirements for machine guarding', true, 14),
('Hearing Conservation', 'OSHA 1910.95 - Occupational noise exposure and hearing conservation', true, 15),
('HAZWOPER', 'OSHA 1910.120 - Hazardous waste operations and emergency response', true, 16)
""")
# ================================================================
# Seed data: Certification types
# ================================================================
op.execute("""
INSERT INTO safepath_certification_types (name, description, default_validity_months, retraining_frequency_months, osha_standard_ref, is_system) VALUES
('OSHA 10-Hour General Industry', 'OSHA 10-hour outreach training for general industry workers', NULL, 60, '29 CFR 1910', true),
('OSHA 10-Hour Construction', 'OSHA 10-hour outreach training for construction workers', NULL, 60, '29 CFR 1926', true),
('OSHA 30-Hour General Industry', 'OSHA 30-hour outreach training for supervisors/managers in general industry', NULL, 60, '29 CFR 1910', true),
('OSHA 30-Hour Construction', 'OSHA 30-hour outreach training for supervisors/managers in construction', NULL, 60, '29 CFR 1926', true),
('Forklift Operator', 'Powered industrial truck operator certification', 36, 36, '1910.178', true),
('Respiratory Fit Test', 'Annual respiratory protection fit test certification', 12, 12, '1910.134', true),
('CPR/First Aid', 'CPR and First Aid certification (AHA or Red Cross)', 24, 24, NULL, true),
('AED Certification', 'Automated External Defibrillator training certification', 24, 24, NULL, true),
('Bloodborne Pathogens', 'Annual bloodborne pathogen exposure control training', 12, 12, '1910.1030', true),
('Confined Space Entry', 'Permit-required confined space entry training', NULL, NULL, '1910.146', true),
('HAZWOPER 40-Hour', 'Hazardous waste operations initial 40-hour training', NULL, 12, '1910.120', true),
('HAZWOPER 8-Hour Refresher', 'Annual HAZWOPER refresher training', 12, 12, '1910.120', true),
('Fall Protection Competent Person', 'Competent person designation for fall protection', NULL, NULL, '1926.503', true),
('Scaffolding Competent Person', 'Competent person designation for scaffolding', NULL, NULL, '1926.451', true),
('Fire Extinguisher Training', 'Annual portable fire extinguisher training', 12, 12, '1910.157', true),
('Hearing Conservation', 'Annual hearing conservation program training', 12, 12, '1910.95', true),
('Lockout/Tagout Authorized', 'Authorized employee LOTO training', NULL, 12, '1910.147', true),
('Electrical Safety Qualified', 'Qualified electrical worker training', NULL, NULL, '1910.331', true),
('Crane/Hoist Operator', 'Crane and hoist operator certification', 36, 36, '1926.1427', true),
('Excavation Competent Person', 'Competent person designation for excavation/trenching', NULL, NULL, '1926.651', true)
""")
def downgrade() -> None:
op.drop_table('safepath_audit_log')
op.drop_table('safepath_auto_assignment_rules')
op.drop_table('safepath_classroom_sessions')
op.drop_table('safepath_certifications')
op.drop_table('safepath_results')
op.drop_table('safepath_assignments')
op.drop_table('safepath_quiz_questions')
op.drop_table('safepath_quizzes')
op.drop_table('safepath_lesson_assets')
op.drop_table('safepath_lessons')
op.drop_table('safepath_courses')
op.drop_table('safepath_certification_types')
op.drop_table('safepath_course_categories')
SQLAlchemy Models
File: tellus-ehs-hazcom-service/app/db/models/safepath.py
"""SafePath Training Models
All database models for the SafePath Training module:
- CourseCategory: Pre-built and custom course categories
- CertificationType: Certification types with validity rules
- Course: Training courses with versioning
- Lesson: Individual lessons within a course
- LessonAsset: File assets (PDF, video) attached to lessons
- Quiz: Quizzes attached to courses
- QuizQuestion: Individual quiz questions
- Assignment: Training assignments to users
- Result: Training completion results and quiz scores
- Certification: Employee certification records
- ClassroomSession: In-person training session records
- AutoAssignmentRule: Auto-assignment trigger rules
- AuditLog: Audit trail for all SafePath events
"""
from datetime import datetime, date, time as dt_time
from typing import Optional, List
from uuid import uuid4
from sqlalchemy import (
Boolean,
BigInteger,
CheckConstraint,
Column,
Date,
DateTime,
ForeignKey,
Integer,
Numeric,
String,
Text,
Time,
)
from sqlalchemy.dialects.postgresql import UUID, JSONB
from sqlalchemy.orm import relationship
from app.db.session import Base
# ============================================================================
# Reference / Lookup Tables
# ============================================================================
class CourseCategory(Base):
"""Course category for organizing training courses.
System categories (is_system=True) are shared across all companies.
Custom categories (is_system=False) are company-specific.
"""
__tablename__ = "safepath_course_categories"
category_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
name = Column(String(100), nullable=False)
description = Column(Text, nullable=True)
is_system = Column(Boolean, default=True, nullable=False)
company_id = Column(
UUID(as_uuid=True),
ForeignKey("core_data_companies.company_id", ondelete="CASCADE"),
nullable=True,
)
sort_order = Column(Integer, default=0, nullable=False)
created_at = Column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
# Relationships
courses = relationship("Course", back_populates="category")
class CertificationType(Base):
"""Certification type with default validity and retraining rules.
System types (is_system=True) are pre-configured OSHA certifications.
Custom types (is_system=False) are company-specific.
"""
__tablename__ = "safepath_certification_types"
type_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
name = Column(String(150), nullable=False)
description = Column(Text, nullable=True)
default_validity_months = Column(Integer, nullable=True)
retraining_frequency_months = Column(Integer, nullable=True)
osha_standard_ref = Column(String(50), nullable=True)
is_system = Column(Boolean, default=True, nullable=False)
company_id = Column(
UUID(as_uuid=True),
ForeignKey("core_data_companies.company_id", ondelete="CASCADE"),
nullable=True,
)
created_at = Column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
# Relationships
certifications = relationship("Certification", back_populates="certification_type")
# ============================================================================
# Course & Content Tables
# ============================================================================
class Course(Base):
"""Training course with versioning support.
Courses go through draft -> published -> archived lifecycle.
When a published course is edited, a new version is created
(parent_course_id points to the original). Existing assignments
stay pinned to their assigned version.
"""
__tablename__ = "safepath_courses"
course_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
company_id = Column(
UUID(as_uuid=True),
ForeignKey("core_data_companies.company_id", ondelete="CASCADE"),
nullable=False,
)
title = Column(String(255), nullable=False)
description = Column(Text, nullable=True)
category_id = Column(
UUID(as_uuid=True),
ForeignKey("safepath_course_categories.category_id", ondelete="SET NULL"),
nullable=True,
)
osha_standard_ref = Column(String(50), nullable=True)
estimated_duration_minutes = Column(Integer, nullable=True)
passing_score_percent = Column(Integer, default=80, nullable=False)
max_retakes = Column(Integer, default=3, nullable=False)
status = Column(String(20), default="draft", nullable=False)
version_number = Column(Integer, default=1, nullable=False)
parent_course_id = Column(
UUID(as_uuid=True),
ForeignKey("safepath_courses.course_id", ondelete="SET NULL"),
nullable=True,
)
plan_version_id = Column(UUID(as_uuid=True), nullable=True)
locale = Column(String(5), default="en", nullable=False)
created_by = Column(
UUID(as_uuid=True),
ForeignKey("users.user_id", ondelete="SET NULL"),
nullable=True,
)
created_at = Column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime(timezone=True), onupdate=datetime.utcnow, nullable=True)
# Relationships
category = relationship("CourseCategory", back_populates="courses")
lessons = relationship("Lesson", back_populates="course", cascade="all, delete-orphan", order_by="Lesson.sort_order")
quizzes = relationship("Quiz", back_populates="course", cascade="all, delete-orphan", order_by="Quiz.sort_order")
assignments = relationship("Assignment", back_populates="course")
classroom_sessions = relationship("ClassroomSession", back_populates="course")
parent_course = relationship("Course", remote_side=[course_id], foreign_keys=[parent_course_id])
auto_assignment_rules = relationship("AutoAssignmentRule", back_populates="course")
__table_args__ = (
CheckConstraint("status IN ('draft', 'published', 'archived')", name='ck_safepath_courses_status'),
CheckConstraint('passing_score_percent >= 0 AND passing_score_percent <= 100', name='ck_safepath_courses_passing_score'),
CheckConstraint('max_retakes >= 0', name='ck_safepath_courses_max_retakes'),
)
class Lesson(Base):
"""Individual lesson within a course.
Supports multiple content types: video, pdf, slides, text, external_link.
Content is stored as JSONB to accommodate different structures per type.
"""
__tablename__ = "safepath_lessons"
lesson_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
course_id = Column(
UUID(as_uuid=True),
ForeignKey("safepath_courses.course_id", ondelete="CASCADE"),
nullable=False,
)
title = Column(String(255), nullable=False)
lesson_type = Column(String(20), nullable=False)
content = Column(JSONB, nullable=True)
sort_order = Column(Integer, default=0, nullable=False)
completion_threshold_percent = Column(Integer, default=80, nullable=False)
locale = Column(String(5), default="en", nullable=False)
created_at = Column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime(timezone=True), onupdate=datetime.utcnow, nullable=True)
# Relationships
course = relationship("Course", back_populates="lessons")
assets = relationship("LessonAsset", back_populates="lesson", cascade="all, delete-orphan")
__table_args__ = (
CheckConstraint(
"lesson_type IN ('video', 'pdf', 'slides', 'text', 'external_link')",
name='ck_safepath_lessons_type',
),
)
class LessonAsset(Base):
"""File asset attached to a lesson (PDF, video, etc.)."""
__tablename__ = "safepath_lesson_assets"
asset_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
lesson_id = Column(
UUID(as_uuid=True),
ForeignKey("safepath_lessons.lesson_id", ondelete="CASCADE"),
nullable=False,
)
file_name = Column(String(255), nullable=False)
s3_bucket = Column(String(100), nullable=False)
s3_key = Column(Text, nullable=False)
file_type = Column(String(50), nullable=False)
file_size_bytes = Column(BigInteger, nullable=True)
created_at = Column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
# Relationships
lesson = relationship("Lesson", back_populates="assets")
# ============================================================================
# Quiz Tables
# ============================================================================
class Quiz(Base):
"""Quiz attached to a course."""
__tablename__ = "safepath_quizzes"
quiz_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
course_id = Column(
UUID(as_uuid=True),
ForeignKey("safepath_courses.course_id", ondelete="CASCADE"),
nullable=False,
)
title = Column(String(255), default="Course Quiz", nullable=False)
sort_order = Column(Integer, default=0, nullable=False)
created_at = Column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime(timezone=True), onupdate=datetime.utcnow, nullable=True)
# Relationships
course = relationship("Course", back_populates="quizzes")
questions = relationship("QuizQuestion", back_populates="quiz", cascade="all, delete-orphan", order_by="QuizQuestion.sort_order")
class QuizQuestion(Base):
"""Individual quiz question.
Options are stored as JSONB array:
[
{"id": "a", "text": "Option A", "is_correct": true},
{"id": "b", "text": "Option B", "is_correct": false},
...
]
For matching type questions:
[
{"id": "1", "left": "GHS01", "right": "Exploding Bomb", "is_correct": true},
...
]
"""
__tablename__ = "safepath_quiz_questions"
question_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
quiz_id = Column(
UUID(as_uuid=True),
ForeignKey("safepath_quizzes.quiz_id", ondelete="CASCADE"),
nullable=False,
)
question_type = Column(String(20), nullable=False)
question_text = Column(Text, nullable=False)
options = Column(JSONB, nullable=False)
explanation = Column(Text, nullable=True)
sort_order = Column(Integer, default=0, nullable=False)
locale = Column(String(5), default="en", nullable=False)
created_at = Column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
# Relationships
quiz = relationship("Quiz", back_populates="questions")
__table_args__ = (
CheckConstraint(
"question_type IN ('mcq_single', 'mcq_multi', 'true_false', 'matching')",
name='ck_safepath_questions_type',
),
)
# ============================================================================
# Assignment & Results Tables
# ============================================================================
class Assignment(Base):
"""Training assignment linking a course to a user.
Tracks assignment lifecycle: pending -> in_progress -> completed/overdue/expired.
Pinned to a specific course version at creation time.
"""
__tablename__ = "safepath_assignments"
assignment_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
company_id = Column(
UUID(as_uuid=True),
ForeignKey("core_data_companies.company_id", ondelete="CASCADE"),
nullable=False,
)
course_id = Column(
UUID(as_uuid=True),
ForeignKey("safepath_courses.course_id", ondelete="CASCADE"),
nullable=False,
)
course_version = Column(Integer, nullable=False)
assigned_to = Column(
UUID(as_uuid=True),
ForeignKey("users.user_id", ondelete="CASCADE"),
nullable=False,
)
assigned_by = Column(
UUID(as_uuid=True),
ForeignKey("users.user_id", ondelete="SET NULL"),
nullable=True,
)
site_id = Column(
UUID(as_uuid=True),
ForeignKey("company_sites.site_id", ondelete="SET NULL"),
nullable=True,
)
due_date = Column(Date, nullable=False)
priority = Column(String(10), default="normal", nullable=False)
status = Column(String(20), default="pending", nullable=False)
notes = Column(Text, nullable=True)
auto_rule_id = Column(UUID(as_uuid=True), nullable=True)
created_at = Column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime(timezone=True), onupdate=datetime.utcnow, nullable=True)
# Relationships
course = relationship("Course", back_populates="assignments")
results = relationship("Result", back_populates="assignment", cascade="all, delete-orphan")
__table_args__ = (
CheckConstraint("status IN ('pending', 'in_progress', 'completed', 'overdue', 'expired')", name='ck_safepath_assignments_status'),
CheckConstraint("priority IN ('normal', 'urgent')", name='ck_safepath_assignments_priority'),
)
class Result(Base):
"""Training completion result for a single attempt.
Records are append-only (immutable). Each quiz attempt creates a new result.
The final passing result triggers certificate generation and assignment completion.
"""
__tablename__ = "safepath_results"
result_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
assignment_id = Column(
UUID(as_uuid=True),
ForeignKey("safepath_assignments.assignment_id", ondelete="CASCADE"),
nullable=False,
)
user_id = Column(
UUID(as_uuid=True),
ForeignKey("users.user_id", ondelete="CASCADE"),
nullable=False,
)
course_id = Column(
UUID(as_uuid=True),
ForeignKey("safepath_courses.course_id", ondelete="CASCADE"),
nullable=False,
)
attempt_number = Column(Integer, nullable=False)
score_percent = Column(Numeric(5, 2), nullable=True)
passed = Column(Boolean, nullable=True)
started_at = Column(DateTime(timezone=True), nullable=True)
completed_at = Column(DateTime(timezone=True), nullable=True)
duration_seconds = Column(Integer, nullable=True)
delivery_method = Column(String(20), default="online", nullable=False)
instructor_name = Column(String(255), nullable=True)
quiz_answers = Column(JSONB, nullable=True)
lesson_progress = Column(JSONB, nullable=True)
acknowledgment_signature = Column(Text, nullable=True)
acknowledgment_ip_hash = Column(String(64), nullable=True)
acknowledgment_at = Column(DateTime(timezone=True), nullable=True)
created_at = Column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
# Relationships
assignment = relationship("Assignment", back_populates="results")
certifications = relationship("Certification", back_populates="result")
__table_args__ = (
CheckConstraint("delivery_method IN ('online', 'in_person', 'external')", name='ck_safepath_results_delivery'),
CheckConstraint('attempt_number >= 1', name='ck_safepath_results_attempt'),
)
# ============================================================================
# Certification Table
# ============================================================================
class Certification(Base):
"""Employee certification record.
Tracks both internal (earned via SafePath courses) and external
certifications (logged manually or uploaded).
"""
__tablename__ = "safepath_certifications"
certification_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
company_id = Column(
UUID(as_uuid=True),
ForeignKey("core_data_companies.company_id", ondelete="CASCADE"),
nullable=False,
)
user_id = Column(
UUID(as_uuid=True),
ForeignKey("users.user_id", ondelete="CASCADE"),
nullable=False,
)
certification_type_id = Column(
UUID(as_uuid=True),
ForeignKey("safepath_certification_types.type_id", ondelete="RESTRICT"),
nullable=False,
)
issuing_authority = Column(String(255), nullable=True)
certification_number = Column(String(100), nullable=True)
issue_date = Column(Date, nullable=False)
expiration_date = Column(Date, nullable=True)
status = Column(String(20), default="active", nullable=False)
evidence_s3_bucket = Column(String(100), nullable=True)
evidence_s3_key = Column(Text, nullable=True)
source = Column(String(20), default="external", nullable=False)
result_id = Column(
UUID(as_uuid=True),
ForeignKey("safepath_results.result_id", ondelete="SET NULL"),
nullable=True,
)
notes = Column(Text, nullable=True)
created_by = Column(
UUID(as_uuid=True),
ForeignKey("users.user_id", ondelete="SET NULL"),
nullable=True,
)
created_at = Column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime(timezone=True), onupdate=datetime.utcnow, nullable=True)
# Relationships
certification_type = relationship("CertificationType", back_populates="certifications")
result = relationship("Result", back_populates="certifications")
__table_args__ = (
CheckConstraint("status IN ('active', 'expiring_soon', 'expired', 'revoked')", name='ck_safepath_certs_status'),
CheckConstraint("source IN ('internal', 'external')", name='ck_safepath_certs_source'),
)
# ============================================================================
# Classroom Session Table
# ============================================================================
class ClassroomSession(Base):
"""In-person/classroom training session.
Attendance is tracked by creating assignments + results with
delivery_method='in_person' for each attendee.
"""
__tablename__ = "safepath_classroom_sessions"
session_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
company_id = Column(
UUID(as_uuid=True),
ForeignKey("core_data_companies.company_id", ondelete="CASCADE"),
nullable=False,
)
course_id = Column(
UUID(as_uuid=True),
ForeignKey("safepath_courses.course_id", ondelete="CASCADE"),
nullable=False,
)
instructor_name = Column(String(255), nullable=False)
session_date = Column(Date, nullable=False)
start_time = Column(Time, nullable=True)
end_time = Column(Time, nullable=True)
location = Column(String(255), nullable=True)
site_id = Column(
UUID(as_uuid=True),
ForeignKey("company_sites.site_id", ondelete="SET NULL"),
nullable=True,
)
max_attendees = Column(Integer, nullable=True)
notes = Column(Text, nullable=True)
created_by = Column(
UUID(as_uuid=True),
ForeignKey("users.user_id", ondelete="SET NULL"),
nullable=True,
)
created_at = Column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
# Relationships
course = relationship("Course", back_populates="classroom_sessions")
# ============================================================================
# Auto-Assignment Rules Table
# ============================================================================
class AutoAssignmentRule(Base):
"""Rule for automatically assigning training based on triggers.
trigger_config JSONB structure varies by trigger_type:
new_hire:
{"role_ids": ["uuid", ...], "site_ids": ["uuid", ...]}
role_change:
{"target_role_ids": ["uuid", ...]}
site_transfer:
{"target_site_ids": ["uuid", ...]}
cert_expiring:
{"certification_type_id": "uuid", "days_before_expiry": 60}
chemical_added:
{"hazard_classes": ["flammable", "corrosive", ...]}
"""
__tablename__ = "safepath_auto_assignment_rules"
rule_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
company_id = Column(
UUID(as_uuid=True),
ForeignKey("core_data_companies.company_id", ondelete="CASCADE"),
nullable=False,
)
course_id = Column(
UUID(as_uuid=True),
ForeignKey("safepath_courses.course_id", ondelete="CASCADE"),
nullable=False,
)
trigger_type = Column(String(30), nullable=False)
trigger_config = Column(JSONB, default=dict, nullable=False)
due_date_offset_days = Column(Integer, default=30, nullable=False)
is_active = Column(Boolean, default=True, nullable=False)
created_by = Column(
UUID(as_uuid=True),
ForeignKey("users.user_id", ondelete="SET NULL"),
nullable=True,
)
created_at = Column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime(timezone=True), onupdate=datetime.utcnow, nullable=True)
# Relationships
course = relationship("Course", back_populates="auto_assignment_rules")
__table_args__ = (
CheckConstraint(
"trigger_type IN ('new_hire', 'role_change', 'site_transfer', 'cert_expiring', 'chemical_added')",
name='ck_safepath_rules_trigger',
),
)
# ============================================================================
# Audit Log Table
# ============================================================================
class SafePathAuditLog(Base):
"""Audit trail for all SafePath events.
event_type values:
course.created, course.published, course.archived, course.updated,
assignment.created, assignment.started, assignment.completed,
quiz.submitted, quiz.passed, quiz.failed,
certificate.issued, certification.logged, certification.expiring,
retraining.triggered, classroom.recorded
"""
__tablename__ = "safepath_audit_log"
log_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
company_id = Column(
UUID(as_uuid=True),
ForeignKey("core_data_companies.company_id", ondelete="CASCADE"),
nullable=False,
)
event_type = Column(String(50), nullable=False)
entity_type = Column(String(30), nullable=False)
entity_id = Column(UUID(as_uuid=True), nullable=False)
user_id = Column(
UUID(as_uuid=True),
ForeignKey("users.user_id", ondelete="SET NULL"),
nullable=True,
)
details = Column(JSONB, nullable=True)
ip_hash = Column(String(64), nullable=True)
created_at = Column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
Register Models
File: tellus-ehs-hazcom-service/app/db/models/__init__.py
Add the following import to ensure Alembic detects the new models:
from app.db.models.safepath import ( # noqa: F401
CourseCategory,
CertificationType,
Course,
Lesson,
LessonAsset,
Quiz,
QuizQuestion,
Assignment,
Result,
Certification,
ClassroomSession,
AutoAssignmentRule,
SafePathAuditLog,
)
JSONB Column Schemas
lesson.content by lesson_type
video:
{
"description": "Watch this video on fall protection requirements",
"video_url": "https://s3.../video.mp4",
"duration_seconds": 600,
"captions_url": "https://s3.../captions.vtt"
}
pdf:
{
"description": "Read the respiratory protection program document",
"page_count": 12
}
text:
{
"body": "<p>Rich text content with <strong>formatting</strong>...</p>"
}
external_link:
{
"url": "https://www.osha.gov/fall-protection",
"description": "OSHA Fall Protection Standards page"
}
slides:
{
"slides": [
{"title": "Slide 1", "body": "Content...", "image_url": null},
{"title": "Slide 2", "body": "Content...", "image_url": "https://s3.../img.png"}
]
}
result.quiz_answers
[
{"question_id": "uuid", "selected": ["a"], "correct": true},
{"question_id": "uuid", "selected": ["b", "c"], "correct": false},
{"question_id": "uuid", "selected": ["true"], "correct": true}
]
result.lesson_progress
{
"lesson-uuid-1": {"completed": true, "percent": 100, "last_position": null},
"lesson-uuid-2": {"completed": false, "percent": 45, "last_position": 270}
}
Verification
After running the migration:
-- Verify all tables created
SELECT table_name FROM information_schema.tables
WHERE table_name LIKE 'safepath_%'
ORDER BY table_name;
-- Expected: 13 tables
-- safepath_assignments
-- safepath_audit_log
-- safepath_auto_assignment_rules
-- safepath_certifications
-- safepath_certification_types
-- safepath_classroom_sessions
-- safepath_course_categories
-- safepath_courses
-- safepath_lesson_assets
-- safepath_lessons
-- safepath_quiz_questions
-- safepath_quizzes
-- safepath_results
-- Verify seed data
SELECT COUNT(*) FROM safepath_course_categories WHERE is_system = true;
-- Expected: 16
SELECT COUNT(*) FROM safepath_certification_types WHERE is_system = true;
-- Expected: 20