"""Access certification — generalized periodic access-review engine. Revision ID: 251 Revises: 250 Create Date: 2026-06-04 Two tables for the IGA access-review primitive (SOC2 CC6.3 / ISO A.5.15 & A.5.18): ``access_certifications`` (one campaign) + ``access_ certification_items`` (one person×entitlement row with its keep/revoke decision). A campaign snapshots entitlements scoped by ``campaign_type`` + ``scope_config`` and assigns each row a reviewer (the holder's manager). PR2 ships the ``software_seat``-by-department builder; the privileged-role review folds into this engine in a follow-up PR. TiDB-FK strictness: all id/FK columns CHAR(36); FKs to identity_persons use ``id_type_for``; tables inherit identity_persons' charset/collation via ``table_kw_for``. DROP-if-VARCHAR pre-flight for re-run convergence. """ from __future__ import annotations import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for revision = "251" down_revision = "250" branch_labels = None depends_on = None _TABLES = ["access_certification_items", "access_certifications"] def _table_exists(name: str) -> bool: return op.get_bind().execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n LIMIT 1" ), {"n": name}, ).scalar() is not None def _id_is_varchar(name: str) -> bool: row = op.get_bind().execute( sa.text( "SELECT DATA_TYPE FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n " "AND COLUMN_NAME = 'id' LIMIT 1" ), {"n": name}, ).scalar() return bool(row) and str(row).lower() == "varchar" def _ts_cols(): return ( 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), ) def upgrade() -> None: persons_id = id_type_for("identity_persons") table_kw = table_kw_for("identity_persons") for name in _TABLES: if _id_is_varchar(name): op.execute(f"DROP TABLE IF EXISTS `{name}`") if not _table_exists("access_certifications"): op.create_table( "access_certifications", sa.Column("id", sa.CHAR(36), nullable=False), sa.Column("label", sa.String(128), nullable=False), sa.Column("campaign_type", sa.String(32), nullable=False), sa.Column("scope_config", sa.JSON(), nullable=True), sa.Column("status", sa.String(16), nullable=False, server_default=sa.text("'in_progress'")), sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), sa.Column("due_at", sa.DateTime(timezone=True), nullable=False), sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), sa.Column("created_by_person_id", persons_id, nullable=True), sa.Column("snapshot_item_count", sa.Integer(), nullable=False, server_default=sa.text("0")), sa.Column("decided_count", sa.Integer(), nullable=False, server_default=sa.text("0")), sa.Column("revoked_count", sa.Integer(), nullable=False, server_default=sa.text("0")), sa.Column("notes", sa.Text(), nullable=True), *_ts_cols(), sa.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint( ["created_by_person_id"], ["identity_persons.id"], ondelete="SET NULL", name="fk_access_cert_creator", ), **table_kw, ) op.create_index("idx_access_cert_status", "access_certifications", ["status"]) op.create_index("idx_access_cert_type", "access_certifications", ["campaign_type"]) op.create_index("idx_access_cert_due", "access_certifications", ["due_at"]) if not _table_exists("access_certification_items"): op.create_table( "access_certification_items", sa.Column("id", sa.CHAR(36), nullable=False), sa.Column("certification_id", sa.CHAR(36), nullable=False), sa.Column("person_id", persons_id, nullable=True), sa.Column("holder_email", sa.String(255), nullable=True), sa.Column("holder_display_name", sa.String(255), nullable=True), sa.Column("access_type", sa.String(32), nullable=False), sa.Column("access_ref", sa.String(128), nullable=True), sa.Column("access_label", sa.String(255), nullable=False), sa.Column("resource_name", sa.String(255), nullable=True), sa.Column("assigned_reviewer_person_id", persons_id, nullable=True), sa.Column("decided_at_deadline", sa.DateTime(timezone=True), nullable=True), sa.Column("decision", sa.String(16), nullable=False, server_default=sa.text("'pending'")), sa.Column("decided_by_person_id", persons_id, nullable=True), sa.Column("decided_at", sa.DateTime(timezone=True), nullable=True), sa.Column("decision_note", sa.Text(), nullable=True), sa.Column("revocation_status", sa.String(16), nullable=False, server_default=sa.text("'none'")), sa.Column("revocation_completed_at", sa.DateTime(timezone=True), nullable=True), sa.Column("revocation_trace_id", sa.CHAR(36), nullable=True), sa.Column("escalated_at", sa.DateTime(timezone=True), nullable=True), sa.Column("snapshot_at", sa.DateTime(timezone=True), nullable=False), sa.Column("raw_payload", sa.JSON(), nullable=True), *_ts_cols(), sa.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint( ["certification_id"], ["access_certifications.id"], ondelete="CASCADE", name="fk_access_cert_item_cert", ), sa.ForeignKeyConstraint( ["person_id"], ["identity_persons.id"], ondelete="SET NULL", name="fk_access_cert_item_person", ), sa.ForeignKeyConstraint( ["assigned_reviewer_person_id"], ["identity_persons.id"], ondelete="SET NULL", name="fk_access_cert_item_reviewer", ), sa.ForeignKeyConstraint( ["decided_by_person_id"], ["identity_persons.id"], ondelete="SET NULL", name="fk_access_cert_item_decider", ), **table_kw, ) op.create_index("idx_access_cert_item_cert_decision", "access_certification_items", ["certification_id", "decision"]) op.create_index("idx_access_cert_item_person", "access_certification_items", ["person_id"]) op.create_index("idx_access_cert_item_reviewer", "access_certification_items", ["assigned_reviewer_person_id"]) def downgrade() -> None: for name in _TABLES: op.execute(f"DROP TABLE IF EXISTS `{name}`")