|
| 1 | +""" |
| 2 | +Rolling-deploy compatibility checks for Django migrations (MRT7xx). |
| 3 | +
|
| 4 | +Mirrors ``core/compat.py`` (Alembic) but operates on Django migration |
| 5 | +operations. These answer a different question from rollback-safety checks: |
| 6 | + "Can the OLD app version survive while the new migration is live?" |
| 7 | +
|
| 8 | +During a rolling deploy the old app and new schema coexist briefly. Operations |
| 9 | +that break that window are flagged here. |
| 10 | +
|
| 11 | +Coverage note: MRT705 (column type change) is Alembic-only. Django's |
| 12 | +``AlterField`` always carries the full field definition with no reference to the |
| 13 | +previous type, so a type change cannot be detected statically without comparing |
| 14 | +against model state — out of scope for a file scanner. |
| 15 | +""" |
| 16 | + |
| 17 | +from __future__ import annotations |
| 18 | + |
| 19 | +import ast |
| 20 | + |
| 21 | +from ..core.detector import RiskWarning |
| 22 | +from .django_detector import DjangoMigrationAST, _warn |
| 23 | + |
| 24 | + |
| 25 | +def _check_compat_remove_field(m: DjangoMigrationAST) -> list[RiskWarning]: |
| 26 | + """MRT701 — RemoveField (DROP COLUMN) breaks old app instances immediately.""" |
| 27 | + warnings = [] |
| 28 | + for op in m.operations: |
| 29 | + if not isinstance(op, ast.Call) or m.op_name(op) != "RemoveField": |
| 30 | + continue |
| 31 | + model = m.kwarg_str(op, "model_name") or "?" |
| 32 | + field = m.kwarg_str(op, "name") or "?" |
| 33 | + warnings.append( |
| 34 | + _warn( |
| 35 | + m, |
| 36 | + "RemoveField during rolling deploy", |
| 37 | + f"Removing '{model}.{field}' will break old app instances still reading it. " |
| 38 | + "Two-step fix: (1) remove all code references and deploy, then " |
| 39 | + "(2) drop the field in a follow-up migration.", |
| 40 | + "error", |
| 41 | + line=op.lineno, |
| 42 | + code="MRT701", |
| 43 | + ) |
| 44 | + ) |
| 45 | + return warnings |
| 46 | + |
| 47 | + |
| 48 | +def _check_compat_rename_field(m: DjangoMigrationAST) -> list[RiskWarning]: |
| 49 | + """MRT702 — RenameField (RENAME COLUMN) breaks old app instances immediately.""" |
| 50 | + warnings = [] |
| 51 | + for op in m.operations: |
| 52 | + if not isinstance(op, ast.Call) or m.op_name(op) != "RenameField": |
| 53 | + continue |
| 54 | + model = m.kwarg_str(op, "model_name") or "?" |
| 55 | + old_name = m.kwarg_str(op, "old_name") or "?" |
| 56 | + new_name = m.kwarg_str(op, "new_name") or "?" |
| 57 | + warnings.append( |
| 58 | + _warn( |
| 59 | + m, |
| 60 | + "RenameField during rolling deploy", |
| 61 | + f"Renaming '{model}.{old_name}' to '{new_name}' breaks old app instances " |
| 62 | + "referencing the old name. Use expand-contract: add new field, backfill, " |
| 63 | + "update code, then drop the old field.", |
| 64 | + "error", |
| 65 | + line=op.lineno, |
| 66 | + code="MRT702", |
| 67 | + ) |
| 68 | + ) |
| 69 | + return warnings |
| 70 | + |
| 71 | + |
| 72 | +def _check_compat_drop_table(m: DjangoMigrationAST) -> list[RiskWarning]: |
| 73 | + """MRT703 — DeleteModel/RenameModel/AlterModelTable breaks old app immediately.""" |
| 74 | + warnings = [] |
| 75 | + for op in m.operations: |
| 76 | + if not isinstance(op, ast.Call): |
| 77 | + continue |
| 78 | + name = m.op_name(op) |
| 79 | + if name == "DeleteModel": |
| 80 | + model = m.kwarg_str(op, "name") or "?" |
| 81 | + warnings.append( |
| 82 | + _warn( |
| 83 | + m, |
| 84 | + "DeleteModel during rolling deploy", |
| 85 | + f"Deleting model '{model}' drops its table and crashes old app instances " |
| 86 | + "still querying it. Remove all ORM references and deploy first, then " |
| 87 | + "delete the model.", |
| 88 | + "error", |
| 89 | + line=op.lineno, |
| 90 | + code="MRT703", |
| 91 | + ) |
| 92 | + ) |
| 93 | + elif name == "RenameModel": |
| 94 | + old_name = m.kwarg_str(op, "old_name") or "?" |
| 95 | + new_name = m.kwarg_str(op, "new_name") or "?" |
| 96 | + warnings.append( |
| 97 | + _warn( |
| 98 | + m, |
| 99 | + "RenameModel during rolling deploy", |
| 100 | + f"Renaming model '{old_name}' to '{new_name}' renames its table and breaks " |
| 101 | + "old app instances querying the old table. Use expand-contract with an " |
| 102 | + "intermediate compatibility layer.", |
| 103 | + "error", |
| 104 | + line=op.lineno, |
| 105 | + code="MRT703", |
| 106 | + ) |
| 107 | + ) |
| 108 | + elif name == "AlterModelTable": |
| 109 | + model = m.kwarg_str(op, "name") or "?" |
| 110 | + warnings.append( |
| 111 | + _warn( |
| 112 | + m, |
| 113 | + "AlterModelTable during rolling deploy", |
| 114 | + f"Changing the db_table of '{model}' breaks old app instances querying the " |
| 115 | + "old table name. Use expand-contract with an intermediate view or copy.", |
| 116 | + "error", |
| 117 | + line=op.lineno, |
| 118 | + code="MRT703", |
| 119 | + ) |
| 120 | + ) |
| 121 | + return warnings |
| 122 | + |
| 123 | + |
| 124 | +def _check_compat_add_field_not_null(m: DjangoMigrationAST) -> list[RiskWarning]: |
| 125 | + """MRT704 — AddField NOT NULL without a default breaks old app INSERTs.""" |
| 126 | + warnings = [] |
| 127 | + for op in m.operations: |
| 128 | + if not isinstance(op, ast.Call) or m.op_name(op) != "AddField": |
| 129 | + continue |
| 130 | + model = m.kwarg_str(op, "model_name") or "?" |
| 131 | + name = m.kwarg_str(op, "name") or "?" |
| 132 | + field_node = m.field_kwarg(op, "field") |
| 133 | + if not isinstance(field_node, ast.Call): |
| 134 | + continue |
| 135 | + |
| 136 | + null_val: bool | None = None |
| 137 | + has_default = False |
| 138 | + for kw in field_node.keywords: |
| 139 | + if kw.arg == "null" and isinstance(kw.value, ast.Constant): |
| 140 | + null_val = bool(kw.value.value) |
| 141 | + if kw.arg in ("default", "server_default"): |
| 142 | + has_default = True |
| 143 | + |
| 144 | + # Django defaults to null=False. Flag when explicitly or implicitly NOT NULL |
| 145 | + # and no default is supplied. |
| 146 | + if null_val is not True and not has_default: |
| 147 | + warnings.append( |
| 148 | + _warn( |
| 149 | + m, |
| 150 | + "AddField NOT NULL without default during rolling deploy", |
| 151 | + f"Adding NOT NULL field '{model}.{name}' without a default causes INSERT " |
| 152 | + "failures from old app instances that don't supply the value. " |
| 153 | + "Add a default or make it nullable first.", |
| 154 | + "error", |
| 155 | + line=op.lineno, |
| 156 | + code="MRT704", |
| 157 | + ) |
| 158 | + ) |
| 159 | + return warnings |
| 160 | + |
| 161 | + |
| 162 | +_DJANGO_COMPAT_CHECKS = [ |
| 163 | + _check_compat_remove_field, |
| 164 | + _check_compat_rename_field, |
| 165 | + _check_compat_drop_table, |
| 166 | + _check_compat_add_field_not_null, |
| 167 | +] |
| 168 | + |
| 169 | + |
| 170 | +def analyze_django_compat(m: DjangoMigrationAST) -> list[RiskWarning]: |
| 171 | + """Run all rolling-deploy compatibility checks on a single Django migration.""" |
| 172 | + results: list[RiskWarning] = [] |
| 173 | + for check in _DJANGO_COMPAT_CHECKS: |
| 174 | + results.extend(check(m)) |
| 175 | + return results |
0 commit comments