"""Question folders, per-section article notes, and feedback on articles Three tables and one column, kept in one revision because they arrived together: * `question_folders` / `question_folder_questions` — a hand-picked set of questions, and `category_grants.folder_id` so the access tree can name one. A grant over a branch covers whatever is filed there next; a folder covers the list somebody wrote, which is the one grantable shape the tree lacked. * `article_section_notes` — a learner's private note on one section, the same shape as `question_notes`. * `article_feedback` — a report to whoever maintains an article, the same shape as `question_feedback`. `section_id` in the last two is a key inside the `articles.sections` JSON, not a foreign key onto `article_section_index`: those rows are dropped whenever an article is unpublished, and a cascade from them would delete a learner's writing because an educator took an article down for an afternoon. Revision ID: i9f0a1b2c3d4 Revises: h1b2c3d4e5f6 """ import sqlalchemy as sa from alembic import op revision = "i9f0a1b2c3d4" down_revision = "h1b2c3d4e5f6" branch_labels = None depends_on = None TABLES = ("question_folders", "question_folder_questions", "article_section_notes", "article_feedback") def upgrade() -> None: bind = op.get_bind() inspector = sa.inspect(bind) # Guarded table by table rather than all-or-nothing: create_all() runs at # startup and builds whatever is missing, so a box that has already booted # this code may have some of these and not others. present = set(inspector.get_table_names()) if "question_folders" not in present: op.create_table( "question_folders", sa.Column("id", sa.Integer(), primary_key=True, index=True), sa.Column("name", sa.String(length=200), nullable=False), sa.Column("description", sa.Text(), nullable=True), sa.Column("user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True), sa.Column("created_at", sa.DateTime(), server_default=sa.func.now()), sa.Column("updated_at", sa.DateTime(), server_default=sa.func.now()), ) op.create_index("ix_question_folders_user_id", "question_folders", ["user_id"]) if "question_folder_questions" not in present: op.create_table( "question_folder_questions", sa.Column("id", sa.Integer(), primary_key=True, index=True), sa.Column("folder_id", sa.Integer(), sa.ForeignKey("question_folders.id", ondelete="CASCADE"), nullable=False), sa.Column("question_id", sa.Integer(), sa.ForeignKey("questions.id", ondelete="CASCADE"), nullable=False), sa.Column("added_by", sa.Integer(), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True), sa.Column("added_at", sa.DateTime(), server_default=sa.func.now()), sa.UniqueConstraint("folder_id", "question_id", name="uq_folder_question"), ) op.create_index("ix_question_folder_questions_folder_id", "question_folder_questions", ["folder_id"]) op.create_index("ix_question_folder_questions_question_id", "question_folder_questions", ["question_id"]) if "article_section_notes" not in present: op.create_table( "article_section_notes", sa.Column("id", sa.Integer(), primary_key=True, index=True), sa.Column("user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False), sa.Column("article_id", sa.Integer(), sa.ForeignKey("articles.id", ondelete="CASCADE"), nullable=False), sa.Column("section_id", sa.String(length=64), nullable=False), sa.Column("section_title", sa.String(length=300), nullable=True), sa.Column("content", sa.Text(), nullable=False, server_default=""), sa.Column("created_at", sa.DateTime(), server_default=sa.func.now()), sa.Column("updated_at", sa.DateTime(), server_default=sa.func.now()), sa.UniqueConstraint("user_id", "article_id", "section_id", name="uq_article_section_note"), ) op.create_index("ix_article_section_notes_user_id", "article_section_notes", ["user_id"]) op.create_index("ix_article_section_notes_article_id", "article_section_notes", ["article_id"]) if "article_feedback" not in present: op.create_table( "article_feedback", sa.Column("id", sa.Integer(), primary_key=True, index=True), sa.Column("article_id", sa.Integer(), sa.ForeignKey("articles.id", ondelete="CASCADE"), nullable=False), sa.Column("section_id", sa.String(length=64), nullable=True), sa.Column("section_title", sa.String(length=300), nullable=True), sa.Column("user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True), sa.Column("message", sa.Text(), nullable=False), sa.Column("status", sa.String(length=20), nullable=False, server_default="open"), sa.Column("reply", sa.Text(), nullable=True), sa.Column("replied_by", sa.Integer(), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True), sa.Column("replied_at", sa.DateTime(), nullable=True), sa.Column("created_at", sa.DateTime(), server_default=sa.func.now()), ) op.create_index("ix_article_feedback_article_id", "article_feedback", ["article_id"]) op.create_index("ix_article_feedback_section_id", "article_feedback", ["section_id"]) op.create_index("ix_article_feedback_status", "article_feedback", ["status"]) # create_all() only ever creates missing tables, never alters an existing # one, so this column is the part that genuinely needs the migration. columns = {col["name"] for col in sa.inspect(bind).get_columns("category_grants")} if "folder_id" not in columns: op.add_column("category_grants", sa.Column("folder_id", sa.Integer(), nullable=True)) op.create_index("ix_category_grants_folder_id", "category_grants", ["folder_id"]) # SQLite cannot add a foreign key to an existing table; the grant is # cleaned up in application code when a folder is deleted either way. if bind.dialect.name != "sqlite": op.create_foreign_key("fk_category_grants_folder", "category_grants", "question_folders", ["folder_id"], ["id"], ondelete="CASCADE") if bind.dialect.name == "sqlite": return # Both of these were written when there were three dimensions, and neither # is expressed on the model, so create_all() will never notice them. # # The check would reject a grant naming only a folder outright; the unique # index collapses every folder grant for one person onto the same key # (0, 0, 0), so a second folder for the same educator would be rejected as # a duplicate of the first. op.execute("ALTER TABLE category_grants DROP CONSTRAINT IF EXISTS ck_grant_has_a_dimension") op.execute(""" ALTER TABLE category_grants ADD CONSTRAINT ck_grant_has_a_dimension CHECK (category_id IS NOT NULL OR exam_id IS NOT NULL OR tag_id IS NOT NULL OR folder_id IS NOT NULL) """) op.execute("DROP INDEX IF EXISTS uq_grant_dimensions") op.execute(""" CREATE UNIQUE INDEX IF NOT EXISTS uq_grant_dimensions ON category_grants (user_id, COALESCE(category_id, 0), COALESCE(exam_id, 0), COALESCE(tag_id, 0), COALESCE(folder_id, 0)) """) def downgrade() -> None: bind = op.get_bind() if bind.dialect.name != "sqlite": op.execute("DELETE FROM category_grants WHERE folder_id IS NOT NULL") op.execute("ALTER TABLE category_grants DROP CONSTRAINT IF EXISTS ck_grant_has_a_dimension") op.execute(""" ALTER TABLE category_grants ADD CONSTRAINT ck_grant_has_a_dimension CHECK (category_id IS NOT NULL OR exam_id IS NOT NULL OR tag_id IS NOT NULL) """) op.execute("DROP INDEX IF EXISTS uq_grant_dimensions") op.execute(""" CREATE UNIQUE INDEX IF NOT EXISTS uq_grant_dimensions ON category_grants (user_id, COALESCE(category_id, 0), COALESCE(exam_id, 0), COALESCE(tag_id, 0)) """) columns = {col["name"] for col in sa.inspect(bind).get_columns("category_grants")} if "folder_id" in columns: if bind.dialect.name != "sqlite": op.drop_constraint("fk_category_grants_folder", "category_grants", type_="foreignkey") op.drop_index("ix_category_grants_folder_id", table_name="category_grants") op.drop_column("category_grants", "folder_id") present = set(sa.inspect(bind).get_table_names()) for table in reversed(TABLES): if table in present: op.drop_table(table)