"""Add CMDB domain tables. Revision ID: 006 Revises: 005 Create Date: 2026-04-11 """ from alembic import op import sqlalchemy as sa revision = "006" down_revision = "005" branch_labels = None depends_on = None def _table_exists(table: str) -> bool: conn = op.get_bind() result = conn.execute( sa.text( "SELECT COUNT(*) FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :t" ), {"t": table}, ) return result.scalar() > 0 def upgrade() -> None: if not _table_exists("cmdb_provider_accounts"): op.create_table( "cmdb_provider_accounts", sa.Column("id", sa.String(36), nullable=False), sa.Column("provider_type", sa.String(32), nullable=False), sa.Column("account_alias", sa.String(128), nullable=False), sa.Column("description", sa.Text(), nullable=True), sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.text("1")), sa.Column("cred_ciphertext", sa.Text(), nullable=True), sa.Column("cred_nonce", sa.String(32), nullable=True), sa.Column("enabled_capabilities", sa.JSON(), nullable=False), sa.Column("sync_cron", sa.String(64), nullable=True), sa.Column("created_by", sa.String(36), 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(), server_onupdate=sa.func.now(), ), sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("provider_type", "account_alias", name="uq_cmdb_provider_accounts"), ) if not _table_exists("cmdb_sync_jobs"): op.create_table( "cmdb_sync_jobs", sa.Column("id", sa.String(36), nullable=False), sa.Column("provider_account_id", sa.String(36), nullable=False), sa.Column("triggered_by", sa.String(16), nullable=False), sa.Column("triggered_by_user_id", sa.String(36), nullable=True), sa.Column("resource_types", sa.JSON(), nullable=False), sa.Column("status", sa.String(16), nullable=False, server_default=sa.text("'queued'")), sa.Column("capability_results", sa.JSON(), nullable=True), sa.Column("total_upserted", sa.Integer(), nullable=False, server_default=sa.text("0")), sa.Column("total_deleted", sa.Integer(), nullable=False, server_default=sa.text("0")), sa.Column("total_errors", sa.Integer(), nullable=False, server_default=sa.text("0")), sa.Column("error_message", sa.Text(), nullable=True), sa.Column("queued_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), sa.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint( ["provider_account_id"], ["cmdb_provider_accounts.id"], name="fk_cmdb_sync_jobs_provider", ), ) op.create_index("idx_cmdb_sync_jobs_provider", "cmdb_sync_jobs", ["provider_account_id"]) op.create_index("idx_cmdb_sync_jobs_status", "cmdb_sync_jobs", ["status"]) if not _table_exists("cmdb_domains"): op.create_table( "cmdb_domains", sa.Column("id", sa.String(36), nullable=False), sa.Column("provider_account_id", sa.String(36), nullable=False), sa.Column("domain_name", sa.String(253), nullable=False), sa.Column("registrar", sa.String(64), nullable=True), sa.Column("expiry_date", sa.Date(), nullable=True), sa.Column("auto_renew", sa.Boolean(), nullable=False, server_default=sa.text("0")), sa.Column("icp_filing_no", sa.String(64), nullable=True), sa.Column("status", sa.String(32), nullable=False, server_default=sa.text("'active'")), sa.Column("owner_person_id", sa.String(36), nullable=True), sa.Column("owner_raw", sa.String(255), nullable=True), sa.Column("raw_data", sa.JSON(), nullable=True), sa.Column("last_seen_sync_job_id", sa.String(36), nullable=True), sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("last_synced_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("provider_account_id", "domain_name", name="uq_cmdb_domains"), sa.ForeignKeyConstraint( ["provider_account_id"], ["cmdb_provider_accounts.id"], name="fk_cmdb_domains_provider", ), sa.ForeignKeyConstraint( ["owner_person_id"], ["identity_persons.id"], name="fk_cmdb_domains_owner", ondelete="SET NULL", ), ) op.create_index("idx_cmdb_domains_expiry", "cmdb_domains", ["expiry_date"]) if not _table_exists("cmdb_servers"): op.create_table( "cmdb_servers", sa.Column("id", sa.String(36), nullable=False), sa.Column("provider_account_id", sa.String(36), nullable=False), sa.Column("instance_id", sa.String(128), nullable=False), sa.Column("name", sa.String(256), nullable=True), sa.Column("status", sa.String(32), nullable=False), sa.Column("region", sa.String(64), nullable=True), sa.Column("zone", sa.String(64), nullable=True), sa.Column("vcpu", sa.Integer(), nullable=True), sa.Column("memory_mb", sa.Integer(), nullable=True), sa.Column("public_ips", sa.JSON(), nullable=True), sa.Column("private_ips", sa.JSON(), nullable=True), sa.Column("os_type", sa.String(16), nullable=True), sa.Column("image_id", sa.String(128), nullable=True), sa.Column("tags", sa.JSON(), nullable=True), sa.Column("owner_person_id", sa.String(36), nullable=True), sa.Column("owner_raw", sa.String(255), nullable=True), sa.Column("raw_data", sa.JSON(), nullable=True), sa.Column("last_seen_sync_job_id", sa.String(36), nullable=True), sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("last_synced_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("provider_account_id", "instance_id", name="uq_cmdb_servers"), sa.ForeignKeyConstraint( ["provider_account_id"], ["cmdb_provider_accounts.id"], name="fk_cmdb_servers_provider", ), sa.ForeignKeyConstraint( ["owner_person_id"], ["identity_persons.id"], name="fk_cmdb_servers_owner", ondelete="SET NULL", ), ) op.create_index("idx_cmdb_servers_status", "cmdb_servers", ["status"]) op.create_index("idx_cmdb_servers_region", "cmdb_servers", ["region"]) if not _table_exists("cmdb_user_mapping"): op.create_table( "cmdb_user_mapping", sa.Column("id", sa.String(36), nullable=False), sa.Column("cmdb_username", sa.String(255), nullable=False), sa.Column("cmdb_email", sa.String(255), nullable=True), sa.Column("cmdb_is_superuser", sa.Boolean(), nullable=False, server_default=sa.text("0")), sa.Column("helix_person_id", sa.String(36), nullable=True), sa.Column("match_status", sa.String(32), nullable=False, server_default=sa.text("'not_in_helix'")), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("cmdb_username", name="uq_cmdb_user_mapping"), sa.ForeignKeyConstraint( ["helix_person_id"], ["identity_persons.id"], name="fk_cmdb_user_mapping_person", ondelete="SET NULL", ), ) def downgrade() -> None: for table in ( "cmdb_user_mapping", "cmdb_servers", "cmdb_domains", "cmdb_sync_jobs", "cmdb_provider_accounts", ): if _table_exists(table): op.drop_table(table)