"""Drop ``cmdb_cloud_accounts`` — consolidated into the scope/grant model. Revision ID: 053 Revises: 052 Create Date: 2026-04-23 The ``cmdb_cloud_accounts`` table was the original "one IAM user per person per tenant" model with per-row ``offboarding_policy``. Migration 052 introduced the tenant → scope → grant lifecycle that subsumes it: * A per-user IAM login is now a ``cmdb_cloud_access_grants`` row on a ``standalone_iam`` scope (one IAM user per person, multiple policies attached as separate grants). * ``offboarding_policy`` logic moves to the scope level — revoke/ transfer decisions live in the grant lifecycle + offboarding blocker types ``cloud_scope_owner`` / ``cloud_access_grant``. Bedrock auto-grant never wrote this table (it stores AK/SK on ``api_key_assignments.detail.bedrock``), so no application data is lost. We deliberately don't add a data-migration step — the table was empty in production at the time of this change. Downgrade recreates the table with the same shape migration 044 / 045 left it so a rollback is safe; rows written post-rollback would be empty. """ from __future__ import annotations import sqlalchemy as sa from alembic import op revision = "053" down_revision = "052" branch_labels = None depends_on = None def _table_exists(bind, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n", ), {"n": name}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() if _table_exists(bind, "cmdb_cloud_accounts"): # Production safety rail: if someone actually added rows, fail # loudly so they see the message and can migrate them manually. count = bind.execute( sa.text( "SELECT COUNT(*) FROM cmdb_cloud_accounts WHERE deleted_at IS NULL" ) ).scalar() if count and count > 0: raise RuntimeError( f"cmdb_cloud_accounts has {count} active rows; refusing to drop. " "Move them into cmdb_cloud_access_grants manually or " "soft-delete the rows first (set deleted_at) if they're " "abandoned, then re-run the upgrade.", ) op.drop_table("cmdb_cloud_accounts") def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, "cmdb_cloud_accounts"): return op.create_table( "cmdb_cloud_accounts", sa.Column("id", sa.String(36), primary_key=True), sa.Column("provider", sa.String(32), nullable=False), sa.Column("account_id", sa.String(255), nullable=False), sa.Column("display_name", sa.String(255)), sa.Column("owner_person_id", sa.String(36)), sa.Column("integration_id", sa.String(36)), sa.Column( "offboarding_policy", sa.String(32), nullable=False, server_default="sso_cascade", ), sa.Column("offboarding_config", sa.JSON), sa.Column( "status", sa.String(32), nullable=False, server_default="active", ), sa.Column("notes", sa.Text), 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.Column("deleted_at", sa.DateTime(timezone=True)), sa.UniqueConstraint( "provider", "account_id", name="uq_cmdb_cloud_accounts_provider_account", ), )