diff --git a/scripts/check_migration_safety.py b/scripts/check_migration_safety.py index f2acd9194..2c0a08907 100644 --- a/scripts/check_migration_safety.py +++ b/scripts/check_migration_safety.py @@ -145,9 +145,7 @@ def check_not_null_without_default(file_path: Path, upgrade_content: str) -> Lis continue # 检查是否有默认值(server_default 或 default) - has_default = bool( - re.search(r"server_default\s*=", call) or re.search(r"\bdefault\s*=", call) - ) + has_default = bool(re.search(r"server_default\s*=", call) or re.search(r"\bdefault\s*=", call)) if not has_default: # 提取表名和列名 @@ -160,14 +158,10 @@ def check_not_null_without_default(file_path: Path, upgrade_content: str) -> Lis table = table_col_match.group(1) col = table_col_match.group(2) issues.append( - f"{file_path.name}: 新增 NOT NULL 约束无默认值 " - f"(表: {table}, 列: {col})- 旧数据为空时迁移失败" + f"{file_path.name}: 新增 NOT NULL 约束无默认值 " f"(表: {table}, 列: {col})- 旧数据为空时迁移失败" ) else: - issues.append( - f"{file_path.name}: 新增 NOT NULL 约束无默认值 " - "- 旧数据为空时迁移失败" - ) + issues.append(f"{file_path.name}: 新增 NOT NULL 约束无默认值 " "- 旧数据为空时迁移失败") return issues @@ -210,9 +204,7 @@ def check_large_table_alter(file_path: Path, upgrade_content: str) -> List[str]: ) # 检测大表上的 drop_column - drop_col_pattern = ( - r"op\.drop_column\(\s*['\"]([^'\"]+)['\"]\s*,\s*['\"]([^'\"]+)['\"]" - ) + drop_col_pattern = r"op\.drop_column\(\s*['\"]([^'\"]+)['\"]\s*,\s*['\"]([^'\"]+)['\"]" for match in re.finditer(drop_col_pattern, upgrade_content): table = match.group(1) col = match.group(2) @@ -245,9 +237,7 @@ def get_new_migrations_via_diff(diff_target: str) -> List[Path]: 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()}") @@ -255,9 +245,7 @@ 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 > 全部 @@ -320,9 +308,7 @@ 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"), @@ -388,9 +374,7 @@ 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: