"""TiDB-parity guard. Hard CI invariant for three "passes on local MySQL, explodes on TiDB / prod" bug classes — each one has bitten this repo and each has a memory note: 1. **Hand-rolled 36-character ForeignKey column types** (TiDB error 3780). This includes ``String(36)``, bare ``CHAR(36)``, and ``GUID()`` columns connected later by a table-level ``ForeignKeyConstraint``. TiDB rejects a CHAR↔VARCHAR FK that MySQL 8 silently accepts: ``identity_persons.id`` / ``asset_software_catalog.id`` are CHAR(36) on prod TiDB but VARCHAR(36) under local CI MySQL, so a child column declared ``sa.String(36)`` (→ VARCHAR) that points at one of them passes locally and fails the migration on prod. The fix is the shared helper ``app.core.migration_helpers.id_type_for(parent)`` which reads information_schema and returns CHAR(36)/String(36) to match the parent. This guard rejects any NEW ``sa.Column(...)`` in a migration that carries BOTH a ``String(36)``/``String(length=36)`` type AND a ``ForeignKey(...)`` in the same column call. Non-FK String(36) columns (~90% of them) are legitimate and NOT flagged. 1b. **Bare ``CHAR(36)`` as a ForeignKey column type** — right data type, wrong *collation*, same error 3780. A column declared without an explicit ``COLLATE`` inherits the TABLE default, which is fine only while every FK in that table points at parents sharing one collation. It stops being fine when a table mixes parents: migration 437 (2026-07-29) took its table default from ``cmdb_cloud_access_grants`` (``utf8mb4_bin`` on prod) and also FK'd to ``identity_persons`` (``utf8mb4_unicode_ci``), and the deploy died on ``fk_3`` after the previous migration had already committed. Local MySQL cannot reproduce it — one collation everywhere. ``id_type_for`` now returns the parent's collation along with its type, so the fix is again "ask the helper". ``sa.CHAR(36, collation=...)`` is explicit and NOT flagged; a bare ``CHAR(36)`` on a non-FK column (a PK is nobody's child) is NOT flagged. 1c. **``sa.Uuid()`` as a DDL column type in a migration** (same error 3780, plus a silently-narrow column even without an FK). SQLAlchemy's ``Uuid`` is CHAR(32) on MySQL; this repo's models mean CHAR(36) by ``Uuid``, which is an alias — ``from app.core.sql_types import GUID as Uuid``. The name therefore reads correctly in a model and betrays you in a migration, where ``sa`` is the real SQLAlchemy. Migration 612 (2026-09-05) did exactly that and wedged the prod chain at 611. Its own tests passed because they build the schema from ORM metadata, where parent and child are both the alias and agree; only the real chain has a CHAR(36) parent from an older migration. Flagged inside ``op.create_table`` / ``op.add_column`` regardless of FK — a column narrower than every query against it is already wrong. A lightweight ``sa.Table(...)`` for a bulk INSERT (migration 545) emits no DDL and is not flagged. 2. **``.nullslast()`` / ``.nullsfirst()``** (MySQL/TiDB error 1064). Neither engine supports ``NULLS LAST`` syntax; SQLAlchemy emits it verbatim. Emulate with ``order_by(col.is_(None), col.desc())``. Currently 0 occurrences — pure prevention. 3. **``func.timestampdiff("MINUTE", ...)`` with a bare string-literal unit** (error 1064). A Python string as the first arg binds as a ``%s`` parameter, producing ``TIMESTAMPDIFF(%s, ...)`` which MySQL/TiDB reject. The unit must be a raw SQL keyword via ``literal_column("MINUTE")``. This guard flags ``timestampdiff(`` immediately followed by a quote; ``literal_column(...)`` and variable forms are allowed. Scope: kind (1) scans ``alembic/versions/`` only; kinds (2) + (3) scan ``app/``. Inline exemption: add ``# tidb-parity-exempt: `` on the offending line or one of the two lines above it (for a String(36)-FK column the marker may sit anywhere within the column's line span or the two lines above its start). The marker forces a human to justify the exception instead of silently re-growing the bug surface. Runs without a database connection — pure text scan. Exits 0 on success, 1 with a human-readable diagnostic on failure. ``scan_text`` is exposed for unit tests. """ from __future__ import annotations import re import sys from pathlib import Path _EXEMPT_MARKER = "tidb-parity-exempt:" # ── kind (1): String(36) + ForeignKey in the same sa.Column(...) ── _STRING36 = re.compile(r"\bString\(\s*(?:length\s*=\s*)?36\s*\)") # ── kind (1b): bare CHAR(36) + ForeignKey ── # Right data type, WRONG collation. A column with no explicit ``COLLATE`` # inherits the *table* default, and 3780 fires on a collation mismatch just # as readily as on a CHAR↔VARCHAR one. Harmless while every FK in the table # points at parents sharing one collation; a latent deploy failure the moment # the table mixes parents — which is how migration 437 died on 2026-07-29 # (table default utf8mb4_bin from cmdb_cloud_access_grants, person_id FK to # identity_persons' utf8mb4_unicode_ci). ``id_type_for`` now returns the # parent's collation along with its type, so the fix is the same one word: # ask the helper. ``sa.CHAR(36, collation=...)`` is accepted (explicit), as # is a bare CHAR(36) on a NON-FK column (a PK is nobody's child). _CHAR36_BARE = re.compile(r"\bCHAR\(\s*36\s*\)") # ── kind (1c): ``sa.Uuid()`` as a DDL column type ── # SQLAlchemy's own ``Uuid`` renders CHAR(32) on MySQL. In this repo that is # never the intended type: every model spells it # ``from app.core.sql_types import GUID as Uuid``, which is CHAR(36), and # ``identity_persons.id`` is CHAR(36) on prod. So the alias reads correctly in # a model and betrays you in a migration, where ``sa`` is the real SQLAlchemy — # ``sa.Uuid()`` there produces a column four characters narrower than the one # the ORM will query and, against a CHAR(36) parent, an FK MySQL/TiDB reject # outright (errno 3780). Migration 612 (2026-09-05) shipped exactly that and # wedged the prod chain at 611; every later migration was blocked behind it. # # Its tests passed because they build the schema from ORM metadata, where # parent and child are BOTH the CHAR(36) alias and therefore agree — the # mismatch only exists in the real chain, where the parent's CHAR(36) came # from an older migration. That is why this is a text guard and not a test. # # Unlike kinds (1)/(1b), FK or not is irrelevant: a CHAR(32) column the models # address as CHAR(36) is already wrong. Scoped to DDL (``op.create_table`` / # ``op.add_column``) so a lightweight ``sa.Table(...)`` built for a bulk INSERT # — migration 545 — is not flagged: it emits no DDL and defines no type. _SA_UUID = re.compile(r"\bsa\.Uuid\(") _FOREIGN_KEY = re.compile(r"\bForeignKey\(") _SA_COLUMN = re.compile(r"\bsa\.Column\(") _CREATE_TABLE = re.compile(r"\bop\.create_table\(") _OP_EXECUTE = re.compile(r"op\.execute\s*\(") _ADD_COLUMN = re.compile(r"\bop\.add_column\(") _GUID = re.compile(r"\bGUID\(") _FOREIGN_KEY_CONSTRAINT = re.compile(r"\bsa\.ForeignKeyConstraint\(") _QUOTED_NAME = re.compile(r"['\"]([^'\"]+)['\"]") _CREATED_TABLE_NAME = re.compile( r"op\.create_table\(\s*['\"]([a-zA-Z_0-9]+)['\"]" ) # ── kind (2): NULLS LAST / NULLS FIRST ── _NULLS = re.compile(r"\.nulls(?:last|first)\(") # ── kind (3): timestampdiff() with a bare string-literal unit ── # Matches ``timestampdiff(`` followed (after optional whitespace) by a quote. # ``func.timestampdiff(minute_unit, ...)`` / ``timestampdiff(literal_column( # "MINUTE"), ...)`` do NOT match (no leading quote). _TIMESTAMPDIFF_STR = re.compile(r"\btimestampdiff\(\s*['\"]") # Migrations that pre-date this guard and legitimately carry a String(36)-typed # FK column. Computed by running the kind-(1) scan over the whole tree at guard # authoring time (2026-06-08); the guard's job is to stop NEW occurrences, not # rewrite immutable history. These all reference parents created as VARCHAR(36) # within the same migration (so locally consistent) OR pre-date id_type_for and # already shipped to prod successfully. TODO(tidb-parity): if a future prod-TiDB # SHOW CREATE TABLE confirms a CHAR(parent)/VARCHAR(child) mismatch on any of # these, fix forward with an ALTER migration (do NOT edit the historical file). _GRANDFATHERED_STRING36_FK = { "011_cmdb_oss.py", "012_cmdb_applications.py", "013_cmdb_app_resources.py", "042_drop_cmdb_applications_and_owner_raw.py", "072_asset_brands_and_models.py", "074_bailian_tables.py", "084_drop_bailian_applications.py", } # Kind (1b), computed 2026-07-29. Both are exactly the SAFE shape: the FK points # at a parent created IN THE SAME MIGRATION, so child and parent share one table # default and no mixed-parent collation split is possible. Both have also been on # prod TiDB for months. The guard's job is to stop the UNSAFE shape — an FK to a # parent from another migration, whose collation this table cannot see. _GRANDFATHERED_CHAR36_FK = { "015_hr_sync_staging.py", # batch_id -> hr_sync_batches (same file) "037_scheduled_jobs.py", # job_id -> scheduled_jobs (same file) } # Table-level-FK variant of kind (1b), computed 2026-08-16 when that scanner # was widened past GUID. Every one of these predates the guard and has been on # prod TiDB for months, so their collations demonstrably line up; the guard's # job is to stop the NEXT one, not to rewrite shipped history. 253's parent is # an f-string table name the scanner reads literally as ``{_CYCLES}``, which is # also a reminder that the check is textual, not semantic. _GRANDFATHERED_TABLE_LEVEL_36_FK = { "006_cmdb_domain.py", "008_cmdb_dns_lb_cdn.py", "224_procurement_budgets_and_asset_link.py", "253_drop_privileged_review_tables.py", } def _server_dir() -> Path: return Path(__file__).resolve().parent.parent def _rel(path: Path, root: Path) -> str: return str(path.relative_to(root)).replace("\\", "/") def _line_of(text: str, idx: int) -> int: """1-based line number of character offset ``idx``.""" return text.count("\n", 0, idx) + 1 def _balanced_call(text: str, open_paren_idx: int) -> tuple[str, int]: """Return ``(call_text, end_idx)`` for the balanced-paren span starting at ``open_paren_idx`` (which must point at ``(``). ``end_idx`` is the offset of the matching ``)``. Tolerates nested calls (``sa.ForeignKey('t.id', ...)``). """ depth = 0 i = open_paren_idx n = len(text) while i < n: c = text[i] if c == "(": depth += 1 elif c == ")": depth -= 1 if depth == 0: return text[open_paren_idx : i + 1], i i += 1 return text[open_paren_idx:], n - 1 def _exempt_window(lines: list[str], start_line: int, end_line: int) -> bool: """True if the exemption marker appears anywhere in ``[start_line-2, end_line]`` (1-based, inclusive). The two-line lead-in lets a long ORM clause carry the marker as a preceding comment without tripping line-length lint.""" lo = max(0, start_line - 3) # 0-based; two lines above start_line hi = end_line # 0-based exclusive end == 1-based inclusive end_line return any(_EXEMPT_MARKER in w for w in lines[lo:hi]) def _scan_bare_type_fk(text: str, type_re: re.Pattern[str]) -> list[int]: """Line numbers of ``sa.Column(...)`` calls carrying BOTH ``type_re`` and a ``ForeignKey(...)`` — i.e. an FK column whose type was hand-rolled instead of taken from ``id_type_for(parent)``.""" lines = text.splitlines() hits: list[int] = [] for m in _SA_COLUMN.finditer(text): paren = text.index("(", m.start()) call, end_idx = _balanced_call(text, paren) if not (type_re.search(call) and _FOREIGN_KEY.search(call)): continue start_line = _line_of(text, m.start()) end_line = _line_of(text, end_idx) if _exempt_window(lines, start_line, end_line): continue hits.append(start_line) return hits def _tables_created_here(text: str) -> set[str]: """Table names this migration file creates itself.""" return set(_CREATED_TABLE_NAME.findall(text)) def _scan_table_fk( text: str, type_re: re.Pattern[str], *, foreign_parent_only: bool, ) -> list[int]: """Find hand-typed child columns linked by table-level FK constraints. The original scanner only noticed inline ``sa.ForeignKey(...)``. Migration 442 used ``GUID()`` columns plus separate ``ForeignKeyConstraint`` calls, so that production-incompatible shape passed the guard — and the fix added a GUID-only scanner beside it. That left a gap between the two, and migration 472 fell straight into it on 2026-08-16: ``sa.CHAR(36)`` columns wired by table-level ``ForeignKeyConstraint``. Neither scanner looked at that combination, so a shape this guard exists to reject reached prod, failed with 3780 on ``fk_2``, and stranded the chain at 471 — every later migration blocked behind it. Hence one parameterised scanner rather than a third bespoke one. ``foreign_parent_only`` is the risk criterion for the 36-char types, the same one ``_GRANDFATHERED_CHAR36_FK`` states in prose: a bare ``CHAR(36)``/``String(36)`` child inherits its own table's default, which is *correct by construction* when the parent is created in this same file (one table default covers both). It is only a latent 3780 when the parent lives in another migration, whose collation this file cannot see. GUID passes ``False`` — its width/type mismatch does not depend on where the parent was born. """ lines = text.splitlines() own_tables = _tables_created_here(text) if foreign_parent_only else set() hits: list[int] = [] for table_match in _CREATE_TABLE.finditer(text): paren = text.index("(", table_match.start()) table_call, table_end = _balanced_call(text, paren) typed_columns: set[str] = set() for column_match in _SA_COLUMN.finditer(table_call): column_paren = table_call.index("(", column_match.start()) column_call, _ = _balanced_call(table_call, column_paren) if not type_re.search(column_call): continue name_match = _QUOTED_NAME.search(column_call) if name_match: typed_columns.add(name_match.group(1)) if not typed_columns: continue for fk_match in _FOREIGN_KEY_CONSTRAINT.finditer(table_call): fk_paren = table_call.index("(", fk_match.start()) fk_call, fk_end = _balanced_call(table_call, fk_paren) operands = re.findall(r"\[([^]]*)\]", fk_call) if not operands: continue child_columns = set(_QUOTED_NAME.findall(operands[0])) if not typed_columns.intersection(child_columns): continue if foreign_parent_only: if len(operands) < 2: continue parents = { ref.split(".", 1)[0] for ref in _QUOTED_NAME.findall(operands[1]) } if not parents - own_tables: continue absolute_start = paren + fk_match.start() absolute_end = paren + fk_end start_line = _line_of(text, absolute_start) end_line = _line_of(text, absolute_end) if not _exempt_window(lines, start_line, end_line): hits.append(start_line) if table_end >= len(text): break return hits def _scan_ddl_column_type(text: str, type_re: re.Pattern[str]) -> list[int]: """Line numbers of ``sa.Column(...)`` calls carrying ``type_re`` inside a DDL statement (``op.create_table`` / ``op.add_column``). Deliberately blind to whether the column is an FK: kind (1c)'s defect is the width itself, so a lone PK is just as wrong as a child column. """ lines = text.splitlines() hits: list[int] = [] for ddl in (_CREATE_TABLE, _ADD_COLUMN): for ddl_match in ddl.finditer(text): paren = text.index("(", ddl_match.start()) ddl_call, _ = _balanced_call(text, paren) for column_match in _SA_COLUMN.finditer(ddl_call): column_paren = ddl_call.index("(", column_match.start()) column_call, column_end = _balanced_call(ddl_call, column_paren) if not type_re.search(column_call): continue start_line = _line_of(text, paren + column_match.start()) end_line = _line_of(text, paren + column_end) if _exempt_window(lines, start_line, end_line): continue hits.append(start_line) return sorted(hits) def _scan_line_pattern(text: str, pattern: re.Pattern[str]) -> list[int]: lines = text.splitlines() hits: list[int] = [] for idx, line in enumerate(lines): if not pattern.search(line): continue window = lines[max(0, idx - 2) : idx + 1] if any(_EXEMPT_MARKER in w for w in window): continue hits.append(idx + 1) return hits #: A raw ``CREATE TABLE`` inside ``op.execute`` that declares a FOREIGN KEY. #: Matched on the SQL text, not on Python calls, because that is the whole #: point of this kind: raw DDL never reaches ``id_type_for``. _RAW_CREATE_TABLE = re.compile(r"CREATE\s+TABLE", re.IGNORECASE) _RAW_FOREIGN_KEY = re.compile(r"FOREIGN\s+KEY", re.IGNORECASE) def _scan_raw_ddl_fk(text: str) -> list[int]: """Line numbers of ``op.execute`` calls whose SQL creates a table with a FOREIGN KEY. Kind (1d). The three kinds above all inspect ``sa.Column(...)`` calls, so they are structurally blind to a table built from a SQL string — and a SQL string is exactly where the author has to get the child column's type AND collation right by hand, with nothing checking it. Migration 619 shipped this shape briefly: it read the parent's ``column_type`` out of ``information_schema`` and interpolated it, which looks careful and is still wrong, because ``column_type`` carries no collation and a column with no explicit ``COLLATE`` inherits the table default. It applied cleanly on a scratch database created in one statement (uniform collation) and failed with the same errno 3780 on a database built through the real chain — the identical local-vs-prod asymmetry kinds (1b)/(1c) exist for. The fix is never "read the type more carefully"; it is ``op.create_table`` + ``id_type_for`` + ``table_kw_for``, which is the one place that lesson is encoded. """ lines = text.splitlines() hits: list[int] = [] for m in _OP_EXECUTE.finditer(text): paren = text.index("(", m.start()) call, end = _balanced_call(text, paren) if not _RAW_CREATE_TABLE.search(call): continue if not _RAW_FOREIGN_KEY.search(call): # A raw CREATE TABLE with no FK cannot hit 3780. Left alone so # the guard stays about the failure it is named for. continue start_line = _line_of(text, m.start()) end_line = _line_of(text, paren + end) if _exempt_window(lines, start_line, end_line): continue hits.append(start_line) return hits def scan_text(kind: str, text: str) -> list[int]: """Return 1-based line numbers of un-exempted offenders of ``kind`` in ``text``. ``kind`` ∈ {"string36_fk", "nullslast", "timestampdiff"}. Exposed for unit testing: feed a snippet, assert the offending line is caught, the exempted/legitimate forms are not. """ if kind == "string36_fk": return _scan_bare_type_fk(text, _STRING36) if kind == "char36_fk": return _scan_bare_type_fk(text, _CHAR36_BARE) if kind == "raw_ddl_fk": return _scan_raw_ddl_fk(text) if kind == "sa_uuid_ddl": return _scan_ddl_column_type(text, _SA_UUID) if kind == "guid_table_fk": return _scan_table_fk(text, _GUID, foreign_parent_only=False) if kind == "char36_table_fk": return _scan_table_fk(text, _CHAR36_BARE, foreign_parent_only=True) if kind == "string36_table_fk": return _scan_table_fk(text, _STRING36, foreign_parent_only=True) if kind == "nullslast": return _scan_line_pattern(text, _NULLS) if kind == "timestampdiff": return _scan_line_pattern(text, _TIMESTAMPDIFF_STR) raise ValueError(f"unknown kind: {kind!r}") def main() -> int: server = _server_dir() versions = server / "alembic" / "versions" app_dir = server / "app" if not versions.is_dir() or not app_dir.is_dir(): print(f"[tidb-parity-guard] expected dirs not found under {server}") return 1 offenders: list[str] = [] # kind (1): String(36)-typed FK columns in NEW migrations. for path in sorted(versions.glob("*.py")): if path.name in _GRANDFATHERED_STRING36_FK: continue text = path.read_text(encoding="utf-8") for line_no in scan_text("string36_fk", text): offenders.append( f" [String(36)+FK] alembic/versions/{path.name}:{line_no} — " f"use id_type_for() so CHAR/VARCHAR matches the " f"parent id (TiDB error 3780)." ) for line_no in scan_text("raw_ddl_fk", text): offenders.append( f" [raw CREATE TABLE + FK] alembic/versions/{path.name}:" f"{line_no} — a table built from a SQL string never reaches " f"id_type_for, so the child column's collation is whatever " f"the table default happens to be (TiDB/MySQL error 3780 " f"fires on a collation mismatch even when the data types " f"match). Use op.create_table with " f"id_type_for() + table_kw_for()." ) for line_no in scan_text("sa_uuid_ddl", text): offenders.append( f" [sa.Uuid() in DDL] alembic/versions/{path.name}:{line_no} " f"— sa.Uuid() is CHAR(32) on MySQL, but this repo's models " f"mean CHAR(36) by `Uuid` (it is an alias for " f"app.core.sql_types.GUID). Use id_type_for() " f"for a child column, or the same CHAR(36) type as the ORM " f"for a PK — otherwise the column is narrower than every " f"query against it, and an FK to a CHAR(36) parent is " f"rejected outright (TiDB/MySQL error 3780)." ) for line_no in scan_text("guid_table_fk", text): offenders.append( f" [GUID()+table FK] alembic/versions/{path.name}:{line_no} — " f"a GUID child linked by ForeignKeyConstraint inherits its " f"own table type/collation. Mirror the referenced parent " f"metadata before adding the FK (TiDB error 3780)." ) if path.name not in _GRANDFATHERED_TABLE_LEVEL_36_FK: for kind in ("char36_table_fk", "string36_table_fk"): for line_no in scan_text(kind, text): offenders.append( f" [36-char + table-level FK] " f"alembic/versions/{path.name}:{line_no} — the child " f"column is typed by hand and wired by " f"ForeignKeyConstraint, so it inherits THIS table's " f"collation while the parent lives in another " f"migration with a collation this file cannot see " f"(TiDB error 3780). Type it with " f"id_type_for()." ) if path.name in _GRANDFATHERED_CHAR36_FK: continue for line_no in scan_text("char36_fk", text): offenders.append( f" [CHAR(36)+FK] alembic/versions/{path.name}:{line_no} — " f"bare CHAR(36) inherits the TABLE collation; an FK to a " f"parent that collates differently is rejected with TiDB " f"error 3780. Use id_type_for(), which returns " f"the parent's collation too." ) # kinds (2) + (3): app source. for path in sorted(app_dir.rglob("*.py")): text = path.read_text(encoding="utf-8") rel = _rel(path, server) for line_no in scan_text("nullslast", text): offenders.append( f" [nullslast] {rel}:{line_no} — MySQL/TiDB have no NULLS " f"LAST/FIRST (error 1064); emulate with " f"order_by(col.is_(None), col.desc())." ) for line_no in scan_text("timestampdiff", text): offenders.append( f" [timestampdiff] {rel}:{line_no} — a string-literal unit " f"binds as %s (error 1064); wrap it in " f'literal_column("MINUTE").' ) if offenders: print( "[tidb-parity-guard] FAIL — TiDB/prod-incompatible SQL that local " "MySQL would accept. Fix as noted, or add " "'# tidb-parity-exempt: ' on/above the line if it is a " "deliberate, verified exception:\n" + "\n".join(offenders) ) return 1 print( "[tidb-parity-guard] OK — no hand-rolled String(36)/CHAR(36) FK column " "types, no sa.Uuid() in migration DDL, NULLS LAST/FIRST, " "or string-literal timestampdiff units." ) return 0 if __name__ == "__main__": sys.exit(main())