fix(format): black格式化 check_migration_safety.py
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 20s
CI/CD Pipeline / Integration Tests (pull_request) Failing after 17s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 52s
CI/CD Pipeline / Production Browser E2E (pull_request) Failing after 1552h33m56s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 1552h33m56s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 1552h33m57s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Failing after 1552h33m59s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Failing after 1553h5m31s
CI/CD Pipeline / Build Production Runtime Images (pull_request) Failing after 1553h5m33s

This commit is contained in:
ops-bot
2026-07-13 22:52:50 +08:00
parent 93d418fcce
commit 74566b2dfe
+34 -9
View File
@@ -48,10 +48,19 @@ HIGH_RISK_PATTERNS = [
# 中风险模式:可能导致数据丢失或兼容性问题
MEDIUM_RISK_PATTERNS = [
(r"op\.alter_column\([^)]*nullable\s*=\s*False", "新增 NOT NULL 约束 - 旧数据可能为空导致迁移失败"),
(
r"op\.alter_column\([^)]*nullable\s*=\s*False",
"新增 NOT NULL 约束 - 旧数据可能为空导致迁移失败",
),
(r"op\.alter_column\([^)]*type_\s*=", "列类型变更 - 可能导致数据截断或转换失败"),
(r"\bop\.rename_table\(", "op.rename_table() - 重命名表,可能导致依赖该表的代码报错"),
(r"\bop\.rename_column\(", "op.rename_column() - 重命名列,可能导致依赖该列的代码报错"),
(
r"\bop\.rename_table\(",
"op.rename_table() - 重命名表,可能导致依赖该表的代码报错",
),
(
r"\bop\.rename_column\(",
"op.rename_column() - 重命名列,可能导致依赖该列的代码报错",
),
(r"\bop\.drop_index\(", "op.drop_index() - 删除索引,可能影响查询性能"),
(r"\bop\.drop_constraint\(", "op.drop_constraint() - 删除约束,可能影响数据完整性"),
]
@@ -95,13 +104,24 @@ def get_new_migrations_via_diff(diff_target: str) -> List[Path]:
"""
try:
result = subprocess.run(
["git", "diff", "--name-only", "--diff-filter=A", diff_target, "HEAD", "--", "alembic/versions/"],
[
"git",
"diff",
"--name-only",
"--diff-filter=A",
diff_target,
"HEAD",
"--",
"alembic/versions/",
],
cwd=str(REPO_ROOT),
capture_output=True,
text=True,
check=True,
)
files = [line.strip() for line in result.stdout.strip().split("\n") if line.strip()]
files = [
line.strip() for line in result.stdout.strip().split("\n") if line.strip()
]
return [REPO_ROOT / f for f in files]
except subprocess.CalledProcessError as e:
print(f"⚠️ git diff 失败({diff_target}):{e.stderr.strip()}")
@@ -109,7 +129,9 @@ def get_new_migrations_via_diff(diff_target: str) -> List[Path]:
return sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
def find_new_migrations(since_revision: str | None = None, diff_against: str | None = None) -> List[Path]:
def find_new_migrations(
since_revision: str | None = None, diff_against: str | None = None
) -> List[Path]:
"""
找出需要检查的迁移文件。
优先级:diff_against > since_revision > 全部
@@ -161,7 +183,9 @@ def analyze_migration(file_path: Path) -> Tuple[List[str], List[str], List[str]]
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser = argparse.ArgumentParser(
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
)
parser.add_argument(
"--since",
default=os.getenv("MIGRATION_SINCE_REVISION"),
@@ -224,7 +248,9 @@ def main() -> int:
print()
print("=" * 60)
print(f"检查结果:{len(all_safe)} 项安全 / {len(all_medium)} 项中风险 / {len(all_high)} 项高风险")
print(
f"检查结果:{len(all_safe)} 项安全 / {len(all_medium)} 项中风险 / {len(all_high)} 项高风险"
)
print()
if all_high:
@@ -247,4 +273,3 @@ def main() -> int:
if __name__ == "__main__":
sys.exit(main())