"""api_key_release_requests — Helix-internal release approval table. Revision ID: 192 Revises: 191 Create Date: 2026-05-20 Why this exists --------------- PR #408 shipped the API-Key self-service release flow on top of the Feishu approval-template path: employee submits → ApiKeyService talks to ``FeishuApprovalClient.create_instance`` → an approval card lands in the leader's Feishu inbox → leader clicks 同意 → Feishu callback runs ``handle_release_feishu_callback`` → cloud teardown fires. That shape is identical to ``api_key_request`` and required the admin to build a new ``api_key_release`` template in Feishu open-platform AND configure ``flow_key=api_key_release`` in the connector — two extra config steps for every tenant that wants the flow. The user's intent was different: release should follow the same "Helix-internal approval + Feishu PUSH notification (not approval card)" pattern used by :class:`AssetTransferRequest`. The Feishu push just links the manager to the Helix Web UI — the actual approval button lives in Helix. This migration introduces the persistence backbone for that flow. Schema notes ------------ * ``assignment_id`` / ``requester_id`` / ``approver_id`` are UUID FKs. Their type + collation are mirrored from the target columns at runtime via ``_fk_source_col_type`` — production ``api_key_assignments.id`` is ``char(36)`` collated ``utf8mb4_unicode_ci``, and a naive ``sa.CHAR(36)`` inherits the new table's connection-time collation (often ``utf8mb4_0900_ai_ci`` or ``utf8mb4_bin``) which TiDB rejects with ``(3780, "Referencing column ... incompatible.")``. The PR #408 → PR #416 hotfix cycle established this pattern; see the docstring on ``_fk_source_col_type`` for the full failure mode. * ``approver_id`` is nullable + ``ON DELETE SET NULL`` so a request row survives forensically when a leader is hard-deleted (extremely rare — soft-delete is the norm). * ``previous_status`` captures the pre-release status of the assignment row so reject / cancel can roll it back without a JSON predicate scan over ``assignment.detail``. * Three composite indexes cover the three lookup patterns: (requester, status) for "my open releases", (approver, status) for "what's waiting on my decision", (assignment, status) for the cross-state guard ("is there already a pending release for this AK?"). Idempotency ----------- Table creation guarded via ``information_schema.TABLES`` probe (same shape as 191). Safe to re-run. """ from __future__ import annotations import sqlalchemy as sa from alembic import op revision = "192" down_revision = "191" branch_labels = None depends_on = None def _table_exists(bind, table: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t LIMIT 1" ), {"t": table}, ).scalar() is not None def _fk_source_col_type( bind, target_table: str, target_column: str, *, fallback_type: str = "char", fallback_collation: str | None = "utf8mb4_unicode_ci", length: int = 36, ): """Mirror the target FK column's data type + collation. Same helper PR #416 introduced for the 191 expiry-dedup table. We re-derive it here so the migration is self-contained (alembic migrations don't share a helper module, and the cost of a duplicate ten-line function is trivial against the cost of accidentally pulling in a refactored helper that no longer matches the historical contract). Failure mode this guards against: pymysql.err.OperationalError: (3780, "Referencing column 'assignment_id' and referenced column 'id' in foreign key constraint 'fk_...' are incompatible.") Root cause: ``api_key_assignments.id`` lives in prod as ``char(36)`` collated ``utf8mb4_unicode_ci`` (created by ancient migration 020 before the connection default flipped to ``utf8mb4_0900_ai_ci`` on newer MySQL clients / ``utf8mb4_bin`` on TiDB). A new table's ``sa.CHAR(36)`` inherits the connection collation, producing a strict-mode mismatch TiDB rejects on FK creation. """ row = bind.execute( sa.text( "SELECT DATA_TYPE, COLLATION_NAME " "FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() " " AND TABLE_NAME = :t AND COLUMN_NAME = :c" ), {"t": target_table, "c": target_column}, ).fetchone() if row: data_type = (row[0] or fallback_type).lower() collation = row[1] or fallback_collation else: data_type = fallback_type.lower() collation = fallback_collation type_cls = sa.CHAR if data_type == "char" else sa.String if collation: return type_cls(length, collation=collation) return type_cls(length) def upgrade() -> None: bind = op.get_bind() if _table_exists(bind, "api_key_release_requests"): return assignment_id_col = _fk_source_col_type( bind, "api_key_assignments", "id", ) person_id_col = _fk_source_col_type( bind, "identity_persons", "id", ) approver_id_col = _fk_source_col_type( bind, "identity_persons", "id", ) op.create_table( "api_key_release_requests", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column( "assignment_id", assignment_id_col, sa.ForeignKey( "api_key_assignments.id", ondelete="CASCADE", ), nullable=False, ), sa.Column( "requester_id", person_id_col, # ``SET NULL`` not ``CASCADE``: a release request is # forensically interesting *after* the requester is # hard-deleted (e.g. "what self-service flows did this # person initiate in their last week?"). CASCADE would # silently drop the row from compliance reports. Keep # the row alive with a NULL requester_id so the audit # trail survives. Same posture as ``approver_id`` below. # # The column MUST be nullable for ``SET NULL`` to be # valid SQL — TiDB / MySQL ≥ 5.7 reject FK creation if # the column is NOT NULL and the ondelete is SET NULL # (ERROR 1830). The application layer enforces "not null # on insert" via the service constructor — every code # path that inserts an ApiKeyReleaseRequestModel passes # ``requester_id=current_user.id`` (line 838 of service.py). sa.ForeignKey( "identity_persons.id", ondelete="SET NULL", ), nullable=True, ), sa.Column( "approver_id", approver_id_col, sa.ForeignKey( "identity_persons.id", ondelete="SET NULL", ), nullable=True, ), sa.Column("reason_code", sa.String(32), nullable=False), sa.Column("reason_note", sa.Text(), nullable=True), sa.Column( "status", sa.String(24), nullable=False, server_default=sa.text("'pending'"), ), sa.Column("decision_note", sa.Text(), nullable=True), sa.Column("previous_status", sa.String(32), nullable=False), sa.Column( "requested_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "responded_at", sa.DateTime(timezone=True), nullable=True, ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Index( "ix_apikey_release_requests_requester_status", "requester_id", "status", ), sa.Index( "ix_apikey_release_requests_approver_status", "approver_id", "status", ), sa.Index( "ix_apikey_release_requests_assignment_status", "assignment_id", "status", ), ) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, "api_key_release_requests"): op.drop_table("api_key_release_requests")