"""CMDB 网络资产 — modifier 字段 + LB owner FK + legacy 业务列. Revision ID: 173 Revises: 172 Create Date: 2026-05-17 This migration prepares cmdb_servers / cmdb_load_balancers / cmdb_domains / cmdb_dns_records / cmdb_cdn_domains for the network-asset feature parity work that ships alongside it: 1. **"Last-modified" tracking** — every PATCH / batch-annotate / enable / disable / DNS write goes through the service layer which now stamps ``last_modified_by_person_id`` + ``last_modified_at`` on the touched row. We deliberately do NOT reuse ``updated_at`` because the row's ``updated_at`` advances on every cloud-sync upsert (status change, tag refresh, etc.) — surfacing that as "修改人/修改时间" in the UI would mislead users into thinking a sync update was a human edit. 2. **LB owner FK** — ``cmdb_load_balancers`` previously stored owner as a free-text ``owner_raw`` column only. The matching FK pattern was added to CDN in migration 160 / to servers+domains in migration 042. This migration brings LB in line so manager-scope filtering, per-user list filters, and the legacy-CMDB owner backfill all work uniformly across the five resource tables. 3. **Legacy business columns** — the legacy Go CMDB (golang/amber) we are migrating data from carries fields Helix doesn't yet model: ``app`` / ``remark`` on servers + LBs; ``grade`` / ``possessor`` on domain registrations; ``cert_id`` on CDN; ``remark`` everywhere. IDC / bare-metal-specific fields (SN, IPMI, cabinet/rack, model IDs) land in ``raw_data`` JSON, not as proper columns — they only apply to legacy KSY / UCloud rows and live cloud sync never fills them, so promoting them to schema columns would bloat the table. FK column type follows the helper from migration 160 — it mirrors the target column's actual DATA_TYPE so TiDB doesn't reject the constraint with error 3780 (CHAR vs VARCHAR incompatibility). Idempotency: every column / index / FK creation is guarded so a partial apply (a column landed but the FK creation tripped 3780) can re-run cleanly. The backfill UPDATE is also idempotent — it only touches rows where ``owner_person_id IS NULL`` and ``owner_raw`` resolves. """ import sqlalchemy as sa from sqlalchemy import text from alembic import op revision = "173" down_revision = "172" branch_labels = None depends_on = None # ─── helpers (mirror migration 160) ──────────────────────────────── def _fk_source_col_type(bind, target_table: str, target_column: str, *, length: int = 36): 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() data_type = (row[0] or "char").lower() if row else "char" collation = row[1] if row else None type_cls = sa.CHAR if data_type == "char" else sa.String return type_cls(length, collation=collation) if collation else type_cls(length) def _column_exists(bind, table: str, column: str) -> bool: row = bind.execute( sa.text( "SELECT 1 FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() " " AND TABLE_NAME = :t AND COLUMN_NAME = :c" ), {"t": table, "c": column}, ).fetchone() return row is not None def _index_exists(bind, table: str, index: str) -> bool: row = bind.execute( sa.text( "SELECT 1 FROM information_schema.STATISTICS " "WHERE TABLE_SCHEMA = DATABASE() " " AND TABLE_NAME = :t AND INDEX_NAME = :i" ), {"t": table, "i": index}, ).fetchone() return row is not None def _fk_exists(bind, table: str, constraint: str) -> bool: row = bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLE_CONSTRAINTS " "WHERE TABLE_SCHEMA = DATABASE() " " AND TABLE_NAME = :t AND CONSTRAINT_NAME = :c " " AND CONSTRAINT_TYPE = 'FOREIGN KEY'" ), {"t": table, "c": constraint}, ).fetchone() return row is not None def _add_modifier_columns(bind, persons_id_type, table: str) -> None: """Add ``last_modified_by_person_id`` (FK) + ``last_modified_at`` on ``table``. Idempotent.""" fk_col = "last_modified_by_person_id" at_col = "last_modified_at" if not _column_exists(bind, table, fk_col): op.add_column(table, sa.Column(fk_col, persons_id_type, nullable=True)) else: # Repair type drift from a partial apply. op.alter_column(table, fk_col, type_=persons_id_type, existing_nullable=True) if not _column_exists(bind, table, at_col): op.add_column( table, sa.Column(at_col, sa.DateTime(timezone=True), nullable=True), ) idx_name = f"ix_{table}_last_modified_by" if not _index_exists(bind, table, idx_name): op.create_index(idx_name, table, [fk_col]) fk_name = f"fk_{table}_last_modified_by_person_id" if not _fk_exists(bind, table, fk_name): op.create_foreign_key( fk_name, table, "identity_persons", [fk_col], ["id"], ondelete="SET NULL", ) def _add_text_column_if_missing(table: str, column: str, *, length: int | None = None, type_: sa.types.TypeEngine | None = None) -> None: bind = op.get_bind() if _column_exists(bind, table, column): return if type_ is None: type_ = sa.String(length) if length else sa.Text() op.add_column(table, sa.Column(column, type_, nullable=True)) # ─── upgrade ─────────────────────────────────────────────────────── def upgrade() -> None: bind = op.get_bind() persons_id_type = _fk_source_col_type(bind, "identity_persons", "id") # 1) ``last_modified_by_person_id`` + ``last_modified_at`` on all 5 tables. for table in ( "cmdb_servers", "cmdb_domains", "cmdb_dns_records", "cmdb_load_balancers", "cmdb_cdn_domains", ): _add_modifier_columns(bind, persons_id_type, table) # 2) LB owner FK. ``owner_raw`` already exists; add ``owner_person_id`` # in line with the CDN pattern (migration 160) and backfill from # the existing raw text. if not _column_exists(bind, "cmdb_load_balancers", "owner_person_id"): op.add_column( "cmdb_load_balancers", sa.Column("owner_person_id", persons_id_type, nullable=True), ) else: op.alter_column( "cmdb_load_balancers", "owner_person_id", type_=persons_id_type, existing_nullable=True, ) # Backfill from owner_raw (email or display_name match). op.execute(text(""" UPDATE cmdb_load_balancers lb JOIN identity_persons p ON LOWER(p.work_email) = LOWER(lb.owner_raw) SET lb.owner_person_id = CAST(p.id AS CHAR(36)) WHERE lb.owner_raw IS NOT NULL AND lb.owner_raw != '' AND lb.owner_person_id IS NULL AND lb.deleted_at IS NULL AND p.deleted_at IS NULL """)) if not _index_exists(bind, "cmdb_load_balancers", "ix_cmdb_load_balancers_owner_person_id"): op.create_index( "ix_cmdb_load_balancers_owner_person_id", "cmdb_load_balancers", ["owner_person_id"], ) if not _fk_exists(bind, "cmdb_load_balancers", "fk_cmdb_load_balancers_owner_person_id"): op.create_foreign_key( "fk_cmdb_load_balancers_owner_person_id", "cmdb_load_balancers", "identity_persons", ["owner_person_id"], ["id"], ondelete="SET NULL", ) # 3) Legacy business columns. Each table gets a small additive set of # nullable columns that the legacy Go CMDB carries but Helix's # sync-driven schema didn't model. IDC/bare-metal fields (SN, # IPMI, cabinet, rack, switches, model IDs) land in raw_data JSON # instead — they only apply to legacy KSY/UCloud rows and live # sync never fills them. # cmdb_servers: cloud + bare-metal common fields. App + remark are # admin-annotated, the rest may be filled by sync or legacy import. _add_text_column_if_missing("cmdb_servers", "app", length=255) _add_text_column_if_missing("cmdb_servers", "remark", type_=sa.Text()) _add_text_column_if_missing("cmdb_servers", "instance_type", length=64) _add_text_column_if_missing("cmdb_servers", "image_name", length=255) _add_text_column_if_missing("cmdb_servers", "vpc_id", length=128) _add_text_column_if_missing("cmdb_servers", "subnet_id", length=128) _add_text_column_if_missing("cmdb_servers", "project_id", length=128) if not _column_exists(bind, "cmdb_servers", "system_disk_gb"): op.add_column("cmdb_servers", sa.Column("system_disk_gb", sa.Integer(), nullable=True)) if not _column_exists(bind, "cmdb_servers", "data_disk_gb"): op.add_column("cmdb_servers", sa.Column("data_disk_gb", sa.Integer(), nullable=True)) if not _column_exists(bind, "cmdb_servers", "bandwidth_mbps"): op.add_column("cmdb_servers", sa.Column("bandwidth_mbps", sa.Integer(), nullable=True)) if not _column_exists(bind, "cmdb_servers", "expired"): op.add_column( "cmdb_servers", sa.Column("expired", sa.Boolean(), nullable=False, server_default=sa.text("0")), ) # cmdb_load_balancers: legacy LB metadata. _add_text_column_if_missing("cmdb_load_balancers", "app", length=255) _add_text_column_if_missing("cmdb_load_balancers", "remark", type_=sa.Text()) _add_text_column_if_missing("cmdb_load_balancers", "tag", length=255) _add_text_column_if_missing("cmdb_load_balancers", "vpc_id", length=128) _add_text_column_if_missing("cmdb_load_balancers", "subnet_id", length=128) _add_text_column_if_missing("cmdb_load_balancers", "project_id", length=128) if not _column_exists(bind, "cmdb_load_balancers", "listener_count"): op.add_column( "cmdb_load_balancers", sa.Column("listener_count", sa.Integer(), nullable=True), ) # cmdb_domains: registrar lifecycle fields. ``grade`` is a free-form # importance tier set by the registrar admin; ``possessor`` is the # legal registrant name (distinct from owner_person_id which is the # Helix-side person responsible for the domain). _add_text_column_if_missing("cmdb_domains", "grade", length=16) _add_text_column_if_missing("cmdb_domains", "possessor", length=255) _add_text_column_if_missing("cmdb_domains", "remark", type_=sa.Text()) # cmdb_dns_records: only remark (everything else is provider-driven). _add_text_column_if_missing("cmdb_dns_records", "remark", type_=sa.Text()) # cmdb_cdn_domains: legacy cert binding + remark. _add_text_column_if_missing("cmdb_cdn_domains", "cert_id", length=128) _add_text_column_if_missing("cmdb_cdn_domains", "remark", type_=sa.Text()) # ─── downgrade ───────────────────────────────────────────────────── def downgrade() -> None: bind = op.get_bind() # Drop legacy business columns. ``expired`` on cmdb_servers has a # server_default; drop the column outright — no row data we care about # if we're downgrading past 173. for col in ("cert_id", "remark"): if _column_exists(bind, "cmdb_cdn_domains", col): op.drop_column("cmdb_cdn_domains", col) if _column_exists(bind, "cmdb_dns_records", "remark"): op.drop_column("cmdb_dns_records", "remark") for col in ("grade", "possessor", "remark"): if _column_exists(bind, "cmdb_domains", col): op.drop_column("cmdb_domains", col) for col in ("app", "remark", "tag", "vpc_id", "subnet_id", "project_id", "listener_count"): if _column_exists(bind, "cmdb_load_balancers", col): op.drop_column("cmdb_load_balancers", col) for col in ("app", "remark", "instance_type", "image_name", "vpc_id", "subnet_id", "project_id", "system_disk_gb", "data_disk_gb", "bandwidth_mbps", "expired"): if _column_exists(bind, "cmdb_servers", col): op.drop_column("cmdb_servers", col) # Drop LB owner FK. if _fk_exists(bind, "cmdb_load_balancers", "fk_cmdb_load_balancers_owner_person_id"): op.drop_constraint( "fk_cmdb_load_balancers_owner_person_id", "cmdb_load_balancers", type_="foreignkey", ) if _index_exists(bind, "cmdb_load_balancers", "ix_cmdb_load_balancers_owner_person_id"): op.drop_index("ix_cmdb_load_balancers_owner_person_id", table_name="cmdb_load_balancers") if _column_exists(bind, "cmdb_load_balancers", "owner_person_id"): op.drop_column("cmdb_load_balancers", "owner_person_id") # Drop modifier columns from all five tables. for table in ( "cmdb_servers", "cmdb_domains", "cmdb_dns_records", "cmdb_load_balancers", "cmdb_cdn_domains", ): fk_name = f"fk_{table}_last_modified_by_person_id" idx_name = f"ix_{table}_last_modified_by" if _fk_exists(bind, table, fk_name): op.drop_constraint(fk_name, table, type_="foreignkey") if _index_exists(bind, table, idx_name): op.drop_index(idx_name, table_name=table) if _column_exists(bind, table, "last_modified_at"): op.drop_column(table, "last_modified_at") if _column_exists(bind, table, "last_modified_by_person_id"): op.drop_column(table, "last_modified_by_person_id")