fix(ci): make migration graph checker mypy-clean
This commit is contained in:
@@ -49,21 +49,30 @@ def main() -> int:
|
|||||||
return 1
|
return 1
|
||||||
|
|
||||||
try:
|
try:
|
||||||
reachable = {rev.revision for rev in script.walk_revisions(base="base", head="heads") if rev.revision}
|
reachable: set[str] = set()
|
||||||
|
for walk_rev in script.walk_revisions(base="base", head="heads"):
|
||||||
|
if walk_rev is None:
|
||||||
|
continue
|
||||||
|
if walk_rev.revision:
|
||||||
|
reachable.add(walk_rev.revision)
|
||||||
except Exception as exc: # pragma: no cover - CI path
|
except Exception as exc: # pragma: no cover - CI path
|
||||||
print(f"ERROR: failed while walking Alembic revision graph: {exc}")
|
print(f"ERROR: failed while walking Alembic revision graph: {exc}")
|
||||||
return 1
|
return 1
|
||||||
|
|
||||||
all_revisions = {
|
all_revisions: set[str] = set()
|
||||||
rev.revision
|
# Alembic's revision_map is dynamically typed; guard None values.
|
||||||
for rev in script.revision_map._revision_map.values() # type: ignore[attr-defined]
|
for map_rev in script.revision_map._revision_map.values():
|
||||||
if getattr(rev, "revision", None)
|
if map_rev is None:
|
||||||
}
|
continue
|
||||||
|
revision = getattr(map_rev, "revision", None)
|
||||||
|
if revision:
|
||||||
|
all_revisions.add(revision)
|
||||||
|
|
||||||
orphans = sorted(all_revisions - reachable)
|
orphans = sorted(all_revisions - reachable)
|
||||||
if orphans:
|
if orphans:
|
||||||
print("ERROR: orphan Alembic revisions detected (not reachable from heads):")
|
print("ERROR: orphan Alembic revisions detected (not reachable from heads):")
|
||||||
for rev in orphans:
|
for orphan_rev in orphans:
|
||||||
print(f" - {rev}")
|
print(f" - {orphan_rev}")
|
||||||
return 1
|
return 1
|
||||||
|
|
||||||
print("OK: migration graph integrity passed")
|
print("OK: migration graph integrity passed")
|
||||||
|
|||||||
Reference in New Issue
Block a user