pesco-ncr/backend/alembic/versions/0003_capa_fields.py

"""CAPA fields for API Q1 §5.9.1.2 / §6.4.2

Revision ID: 0003
Revises: 0002
Create Date: 2026-08-04

Adds root cause capture, the corrective-action gate (required? +
justification), the corrective action plan (owner + due date),
effectiveness verification, the recurring-issue flag with NCR-to-NCR
links, and the 6M root-cause category lookup (seeded here so upgraded
deployments have the standard values without re-running app.seed).
"""
from alembic import op
import sqlalchemy as sa

revision = "0003"
down_revision = "0002"
branch_labels = None
depends_on = None

MYSQL = {"mysql_charset": "utf8mb4", "mysql_collate": "utf8mb4_unicode_ci"}

SIX_M_CATEGORIES = ["Man", "Machine", "Method", "Material", "Measurement", "Environment"]


def upgrade() -> None:
    op.create_table(
        "root_cause_categories",
        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
        sa.Column("name", sa.String(100), nullable=False, unique=True),
        sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.text("1")),
        **MYSQL,
    )
    categories = sa.table(
        "root_cause_categories", sa.column("name", sa.String), sa.column("is_active", sa.Boolean)
    )
    op.bulk_insert(categories, [{"name": n, "is_active": True} for n in SIX_M_CATEGORIES])

    op.create_table(
        "ncr_links",
        sa.Column(
            "ncr_id", sa.Integer(), sa.ForeignKey("ncrs.id", ondelete="CASCADE"), primary_key=True
        ),
        sa.Column(
            "related_ncr_id",
            sa.Integer(),
            sa.ForeignKey("ncrs.id", ondelete="CASCADE"),
            primary_key=True,
        ),
        **MYSQL,
    )

    op.add_column("ncrs", sa.Column("root_cause", sa.Text(), nullable=True))
    op.add_column(
        "ncrs",
        sa.Column(
            "root_cause_category_id",
            sa.Integer(),
            sa.ForeignKey("root_cause_categories.id"),
            nullable=True,
        ),
    )
    op.add_column("ncrs", sa.Column("corrective_action_required", sa.Boolean(), nullable=True))
    op.add_column(
        "ncrs", sa.Column("corrective_action_justification", sa.Text(), nullable=True)
    )
    op.add_column("ncrs", sa.Column("corrective_action_plan", sa.Text(), nullable=True))
    op.add_column(
        "ncrs",
        sa.Column(
            "corrective_action_owner_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=True
        ),
    )
    op.add_column("ncrs", sa.Column("corrective_action_due_date", sa.Date(), nullable=True))
    op.add_column("ncrs", sa.Column("corrective_action_opened_at", sa.DateTime(), nullable=True))
    op.add_column("ncrs", sa.Column("effectiveness_result", sa.String(20), nullable=True))
    op.add_column("ncrs", sa.Column("effectiveness_notes", sa.Text(), nullable=True))
    op.add_column("ncrs", sa.Column("effectiveness_verified_at", sa.DateTime(), nullable=True))
    op.add_column(
        "ncrs",
        sa.Column(
            "effectiveness_verified_by_id",
            sa.Integer(),
            sa.ForeignKey("users.id"),
            nullable=True,
        ),
    )
    op.add_column(
        "ncrs",
        sa.Column("is_recurring", sa.Boolean(), nullable=False, server_default=sa.text("0")),
    )


def downgrade() -> None:
    for column in (
        "is_recurring",
        "effectiveness_verified_by_id",
        "effectiveness_verified_at",
        "effectiveness_notes",
        "effectiveness_result",
        "corrective_action_opened_at",
        "corrective_action_due_date",
        "corrective_action_owner_id",
        "corrective_action_plan",
        "corrective_action_justification",
        "corrective_action_required",
        "root_cause_category_id",
        "root_cause",
    ):
        op.drop_column("ncrs", column)
    op.drop_table("ncr_links")
    op.drop_table("root_cause_categories")