"""Helpers for alembic migrations to stay portable across MySQL & TiDB. CI MySQL 8 created identity_persons.id / asset_software_catalog.id as VARCHAR(36) utf8mb4_0900_ai_ci. Prod TiDB has them as CHAR(36) utf8mb4_unicode_ci. Migrations that reference these columns need to declare the FK column with the same DATA_TYPE / collation, otherwise MySQL/TiDB rejects the FK with error 3780 ("Referencing column ... are incompatible"). """ from __future__ import annotations import sqlalchemy as sa from alembic import op def _column_meta(table: str, column: str) -> tuple[str | None, str | None]: conn = op.get_bind() row = conn.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": table, "c": column}).fetchone() if not row: return None, None return ((row[0] or "").lower() or None), row[1] def id_type_for(parent_table: str = "identity_persons"): """Return sa.CHAR(36) / sa.String(36) matching the parent id column — **including its collation**. The collation is not cosmetic. Error 3780 fires on a collation mismatch even when the two columns' data types are identical, and a column with no explicit ``COLLATE`` silently inherits the *table* default. That is survivable only while every FK in the table points at parents sharing one collation. It stops being survivable the moment a table mixes parents, which is what broke the 2026-07-29 deploy: ``cmdb_cloud_grant_renewal_requests`` took its table charset from ``cmdb_cloud_access_grants`` (``utf8mb4_bin`` on prod TiDB) and also FKs to ``identity_persons`` (``utf8mb4_unicode_ci``). ``person_id`` inherited ``utf8mb4_bin``, so ``fk_3`` was rejected — on TiDB only, since local MySQL 8 has one collation across both parents and therefore cannot reproduce it. Same root cause as migration 126 (2026-05-07); pinning it here rather than in the calling migration is what makes it non-recurring, because every migration already asks this helper for the type. Falls back to an uncollated type when the parent table doesn't exist yet (fresh DB mid-chain) — there is nothing to match, and the FK the caller is about to declare would fail on the missing table first. """ data_type, collation = _column_meta(parent_table, "id") if data_type == "char": return sa.CHAR(36, collation=collation) return sa.String(36, collation=collation) def table_kw_for(parent_table: str = "identity_persons") -> dict: """Return mysql_charset / mysql_collate matching the parent table. If the parent collation can't be detected (fresh DB created from scratch), return empty so the table inherits the server default. """ _, collation = _column_meta(parent_table, "id") if not collation: return {} charset = collation.split("_", 1)[0] return {"mysql_charset": charset, "mysql_collate": collation} def normalize_fk_column_collation( table: str, column: str, parent_table: str, nullable: bool = True, ) -> None: """MODIFY COLUMN so charset/collation exactly matches parent_table.id. MySQL/TiDB error 3780 fires when an FK is created between two CHAR columns whose collations differ — even if their SQLAlchemy type is identical. A previous alembic run that added the column with ``sa.CHAR(36)`` (no explicit charset) leaves the column using the table's default collation, which often diverges from the parent's (TiDB prod: ``utf8mb4_unicode_ci`` on ``integrations.id`` vs. ``utf8mb4_0900_ai_ci`` on younger tables). This helper is a no-op when: * the parent column has no collation (bigint ids), * the column's current collation already matches the parent's. Otherwise it emits a raw ``ALTER TABLE {table} MODIFY COLUMN {column} CHAR(36) ...`` so the subsequent ``op.create_foreign_key`` succeeds. """ parent_type, parent_collation = _column_meta(parent_table, "id") if not parent_collation: return _, own_collation = _column_meta(table, column) if own_collation == parent_collation: return charset = parent_collation.split("_", 1)[0] sql_type = "CHAR(36)" if parent_type == "char" else "VARCHAR(36)" null_clause = "NULL" if nullable else "NOT NULL" op.get_bind().execute(sa.text( f"ALTER TABLE {table} MODIFY COLUMN {column} {sql_type} " f"CHARACTER SET {charset} COLLATE {parent_collation} {null_clause}" ))