"""Create the provider-neutral cloud resource directory foundation. Revision ID: 529 Revises: 528 """ from __future__ import annotations import json import uuid import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for revision = "529" down_revision = "528" branch_labels = None depends_on = None _JOB = "cmdb.resource_directory_refresh" _INVENTORY_JOB = "cmdb.resource_inventory_refresh" def upgrade() -> None: identity_id = id_type_for("identity_persons") integration_id = id_type_for("integrations") op.create_table( "cmdb_resource_types", sa.Column("id", integration_id, primary_key=True), sa.Column("provider", sa.String(64), nullable=False), sa.Column("resource_type", sa.String(192), nullable=False), sa.Column("service", sa.String(96), nullable=False), sa.Column("category", sa.String(64), nullable=False), sa.Column("display_name", sa.String(192), nullable=False), sa.Column("icon_key", sa.String(64)), sa.Column("sensitivity", sa.String(16), nullable=False, server_default="internal"), sa.Column("schema_version", sa.Integer, nullable=False, server_default="1"), sa.Column("display_schema", sa.JSON), sa.Column("discovered_dynamically", sa.Boolean, nullable=False, server_default=sa.false()), sa.Column("active", sa.Boolean, nullable=False, server_default=sa.true()), sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.UniqueConstraint("provider", "resource_type", name="uq_cmdb_resource_type_provider_kind"), ) op.create_index("ix_cmdb_resource_types_category", "cmdb_resource_types", ["category", "active"]) op.create_table( "cmdb_cloud_resources", sa.Column("id", integration_id, primary_key=True), sa.Column("integration_id", integration_id, nullable=False), sa.Column("resource_type_id", integration_id, nullable=False), sa.Column("provider", sa.String(64), nullable=False), sa.Column("service", sa.String(96), nullable=False), sa.Column("category", sa.String(64), nullable=False), sa.Column("resource_type", sa.String(192), nullable=False), sa.Column("provider_resource_id", sa.Text, nullable=False), sa.Column("resource_identity_hash", sa.String(64), nullable=False), sa.Column("name", sa.String(512), nullable=False), sa.Column("scope_external_id", sa.String(256)), sa.Column("region", sa.String(128)), sa.Column("zone", sa.String(128)), sa.Column("lifecycle_status", sa.String(64)), sa.Column("environment", sa.String(32)), sa.Column("owner_person_id", identity_id), sa.Column("sensitivity", sa.String(16), nullable=False, server_default="internal"), sa.Column("tags", sa.JSON), sa.Column("normalized_metadata", sa.JSON), sa.Column("raw_metadata", sa.JSON), sa.Column("content_hash", sa.String(64), nullable=False), sa.Column("source_table", sa.String(96), nullable=False), sa.Column("source_id", sa.String(36), nullable=False), sa.Column("source_sync_job_id", sa.String(36)), sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("last_changed_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("missing_since", sa.DateTime(timezone=True)), sa.Column("deleted_at", sa.DateTime(timezone=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.ForeignKeyConstraint(["integration_id"], ["integrations.id"], ondelete="CASCADE"), sa.ForeignKeyConstraint(["resource_type_id"], ["cmdb_resource_types.id"]), sa.ForeignKeyConstraint(["owner_person_id"], ["identity_persons.id"], ondelete="SET NULL"), sa.UniqueConstraint("integration_id", "resource_identity_hash", name="uq_cmdb_cloud_resource_identity_hash"), sa.UniqueConstraint("source_table", "source_id", name="uq_cmdb_cloud_resource_source"), ) op.create_index("ix_cmdb_cloud_resources_type_status", "cmdb_cloud_resources", ["resource_type", "lifecycle_status"]) op.create_index("ix_cmdb_cloud_resources_provider_scope", "cmdb_cloud_resources", ["provider", "scope_external_id"]) op.create_index("ix_cmdb_cloud_resources_owner", "cmdb_cloud_resources", ["owner_person_id", "deleted_at"]) op.create_index("ix_cmdb_cloud_resources_seen", "cmdb_cloud_resources", ["last_seen_at", "deleted_at"]) op.create_index("ix_cmdb_cloud_resources_category_region", "cmdb_cloud_resources", ["category", "region"]) op.create_table( "cmdb_cloud_resource_changes", sa.Column("id", integration_id, primary_key=True), sa.Column("resource_id", integration_id, nullable=False), sa.Column("change_type", sa.String(24), nullable=False), sa.Column("changed_fields", sa.JSON), sa.Column("previous_hash", sa.String(64)), sa.Column("current_hash", sa.String(64)), sa.Column("snapshot", sa.JSON), sa.Column("source_sync_job_id", sa.String(36)), sa.Column("observed_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.ForeignKeyConstraint(["resource_id"], ["cmdb_cloud_resources.id"], ondelete="CASCADE"), ) op.create_index("ix_cmdb_resource_changes_resource_at", "cmdb_cloud_resource_changes", ["resource_id", "observed_at"]) op.create_index("ix_cmdb_resource_changes_type_at", "cmdb_cloud_resource_changes", ["change_type", "observed_at"]) bind = op.get_bind() exists = bind.execute(sa.text("SELECT 1 FROM scheduled_jobs WHERE name=:name LIMIT 1"), {"name": _JOB}).scalar() if not exists: bind.execute( sa.text( """ INSERT INTO scheduled_jobs (id, name, display_name, description, handler_key, trigger_type, interval_seconds, cron_expr, timezone, enabled, config, run_count, failure_count, created_at, updated_at) VALUES (:id, :name, :display_name, :description, :name, 'cron', NULL, '7,22,37,52 * * * *', 'Asia/Shanghai', 1, :config, 0, 0, UTC_TIMESTAMP(), UTC_TIMESTAMP()) """ ), { "id": str(uuid.uuid4()), "name": _JOB, "display_name": "云资源目录索引刷新", "description": "每15分钟投影现有CMDB专用表，幂等刷新统一只读资源目录并维护变化历史。", "config": json.dumps({"history_retention_days": 730}), }, ) inventory_exists = bind.execute( sa.text("SELECT 1 FROM scheduled_jobs WHERE name=:name LIMIT 1"), {"name": _INVENTORY_JOB}, ).scalar() if not inventory_exists: bind.execute( sa.text( """ INSERT INTO scheduled_jobs (id, name, display_name, description, handler_key, trigger_type, interval_seconds, cron_expr, timezone, enabled, config, run_count, failure_count, created_at, updated_at) VALUES (:id, :name, :display_name, :description, :name, 'cron', NULL, '17 * * * *', 'Asia/Shanghai', 1, :config, 0, 0, UTC_TIMESTAMP(), UTC_TIMESTAMP()) """ ), { "id": str(uuid.uuid4()), "name": _INVENTORY_JOB, "display_name": "云资源广覆盖发现", "description": "每小时读取显式开启的官方多云只读资源清单；不完整扫描禁止消失清理。", "config": json.dumps({}), }, ) def downgrade() -> None: bind = op.get_bind() bind.execute( sa.text("DELETE FROM scheduled_jobs WHERE name IN (:directory, :inventory)"), {"directory": _JOB, "inventory": _INVENTORY_JOB}, ) op.drop_table("cmdb_cloud_resource_changes") op.drop_table("cmdb_cloud_resources") op.drop_table("cmdb_resource_types")