fix(ci): check_migration_safety.py black格式化修复

.gitea/workflows/ci-cd.yml

@@ -96,6 +96,7 @@ jobs:
           python3 -m flake8 --version
           bandit --version
           pytest --version
+          python3 -m mypy --version
 
       - name: Run code quality checks
         shell: sh
@@ -158,6 +159,12 @@ jobs:
             python3 scripts/check_migration_safety.py --allow-medium-risk
           fi
 
+      - name: Run mypy type check
+        shell: sh
+        run: |
+          set -eu
+          python3 -m mypy apps packages
+
   unit-tests:
     name: Unit Tests
     runs-on: host
@@ -221,6 +228,7 @@ jobs:
           python3 -m pip install -q -r requirements.txt
           python3 -m pip install -q -r requirements-dev.txt
           pytest --version
+          python3 -m mypy --version
 
       - name: Run unit tests with coverage
         shell: sh
@@ -320,6 +328,7 @@ jobs:
           python3 -m pip install -q -r requirements.txt
           python3 -m pip install -q -r requirements-dev.txt
           pytest --version
+          python3 -m mypy --version
 
       - name: Start Redis
         shell: sh

mypy.ini

@@ -0,0 +1,76 @@
+[mypy]
+# 宽松模式：只检查有类型注解的函数和变量
+# 后续逐步收紧配置
+check_untyped_defs = False
+disallow_untyped_defs = False
+disallow_incomplete_defs = False
+
+# 基本配置
+python_version = 3.10
+warn_return_any = False
+warn_unused_ignores = True
+no_implicit_optional = True
+
+# 忽略第三方库和生成代码
+[mypy-alembic.*]
+ignore_missing_imports = True
+
+[mypy-fastapi.*]
+ignore_missing_imports = True
+
+[mypy-sqlalchemy.*]
+ignore_missing_imports = True
+
+[mypy-pydantic.*]
+ignore_missing_imports = True
+
+[mypy-redis.*]
+ignore_missing_imports = True
+
+[mypy-celery.*]
+ignore_missing_imports = True
+
+[mypy-minio.*]
+ignore_missing_imports = True
+
+[mypy-jwt.*]
+ignore_missing_imports = True
+
+[mypy-passlib.*]
+ignore_missing_imports = True
+
+[mypy-ffmpeg.*]
+ignore_missing_imports = True
+
+[mypy-tortoise.*]
+ignore_missing_imports = True
+
+[mypy-uvicorn.*]
+ignore_missing_imports = True
+
+[mypy-starlette.*]
+ignore_missing_imports = True
+
+[mypy-httpx.*]
+ignore_missing_imports = True
+
+[mypy-aiofiles.*]
+ignore_missing_imports = True
+
+[mypy-boto3.*]
+ignore_missing_imports = True
+
+[mypy-botocore.*]
+ignore_missing_imports = True
+
+# 生成代码和迁移文件
+[mypy-alembic.versions.*]
+ignore_errors = True
+
+[mypy-*.migrations.*]
+ignore_errors = True
+
+# 测试文件可以宽松一些
+[mypy-tests.*]
+ignore_errors = False
+warn_return_any = False

requirements-dev.txt

@@ -5,6 +5,7 @@ black==26.5.1
 isort==8.0.1
 flake8==7.3.0
 bandit==1.9.4
+mypy==1.11.0
 
 # 测试
 pytest==8.3.3

scripts/check_migration_safety.py

@@ -3,11 +3,11 @@
 数据库迁移破坏性变更安全检查
 
 只检查 Alembic 迁移文件的 upgrade 函数中是否包含破坏性操作：
-- DROP TABLE
-- ALTER TABLE ... DROP COLUMN
-- 列类型变更（可能导致数据丢失）
-- NOT NULL 约束新增（无默认值时）
-- RENAME TABLE / RENAME COLUMN
+- DROP TABLE / DROP COLUMN（高风险，直接阻断）
+- 不加默认值的 NOT NULL 列添加（中风险，告警）
+- 大表的 ALTER TABLE（超过10万行，中风险，告警）
+- 列类型变更（可能导致数据丢失，中风险）
+- RENAME TABLE / RENAME COLUMN（中风险）
 
 忽略 downgrade 函数中的操作（那是回滚逻辑，正常的）。
 
@@ -40,6 +40,14 @@ from typing import List, Tuple
 REPO_ROOT = Path(__file__).resolve().parents[1]
 ALEMBIC_VERSIONS_DIR = REPO_ROOT / "alembic" / "versions"
 
+# 已知可能的大表（超过10万行的表），ALTER TABLE 这些表需要告警
+# 可通过环境变量 MIGRATION_LARGE_TABLES 覆盖，用逗号分隔
+LARGE_TABLES = os.getenv(
+    "MIGRATION_LARGE_TABLES",
+    "generation_projects,generation_tasks,assets,upload_sessions,users",
+).split(",")
+LARGE_TABLE_THRESHOLD = int(os.getenv("MIGRATION_LARGE_TABLE_THRESHOLD", "100000"))
+
 # 高风险模式：直接导致数据丢失（只在 upgrade 中检查）
 HIGH_RISK_PATTERNS = [
     (r"\bop\.drop_table\(", "op.drop_table() - 删除表，数据永久丢失"),
@@ -47,11 +55,17 @@ HIGH_RISK_PATTERNS = [
 ]
 
 # 中风险模式：可能导致数据丢失或兼容性问题
+# 注意：NOT NULL 检测需要额外判断是否有默认值，使用专用函数检测
 MEDIUM_RISK_PATTERNS = [
-    (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() - 删除约束，可能影响数据完整性"),
 ]
@@ -88,6 +102,119 @@ def extract_upgrade_content(content: str) -> str:
     return content[upgrade_start:upgrade_end]
 
 
+def extract_alter_column_calls(upgrade_content: str) -> List[str]:
+    """
+    从 upgrade 内容中提取所有 op.alter_column() 调用的完整参数字符串。
+    处理跨行的情况。
+    """
+    calls = []
+    pattern = r"op\.alter_column\("
+    pos = 0
+    while True:
+        match = re.search(pattern, upgrade_content[pos:])
+        if not match:
+            break
+        call_start = pos + match.start()
+        # 从 opening 括号之后开始计数，初始深度为 1
+        i = call_start + len("op.alter_column(")
+        depth = 1
+        while i < len(upgrade_content) and depth > 0:
+            if upgrade_content[i] == "(":
+                depth += 1
+            elif upgrade_content[i] == ")":
+                depth -= 1
+            i += 1
+        if depth == 0:
+            # 找到了匹配的闭合括号，i 指向闭合括号之后
+            calls.append(upgrade_content[call_start : i - 1])
+        pos = i
+    return calls
+
+
+def check_not_null_without_default(file_path: Path, upgrade_content: str) -> List[str]:
+    """
+    检测 NOT NULL 列添加但没有默认值的情况。
+    只有 nullable=False 且没有 server_default / default 时才告警。
+    """
+    issues = []
+    alter_calls = extract_alter_column_calls(upgrade_content)
+
+    for call in alter_calls:
+        # 检查是否设置了 nullable=False
+        if not re.search(r"nullable\s*=\s*False", call):
+            continue
+
+        # 检查是否有默认值（server_default 或 default）
+        has_default = bool(re.search(r"server_default\s*=", call) or re.search(r"\bdefault\s*=", call))
+
+        if not has_default:
+            # 提取表名和列名
+            # op.alter_column('table_name', 'column_name', ...)
+            table_col_match = re.search(
+                r"op\.alter_column\(\s*['\"]([^'\"]+)['\"]\s*,\s*['\"]([^'\"]+)['\"]",
+                call,
+            )
+            if table_col_match:
+                table = table_col_match.group(1)
+                col = table_col_match.group(2)
+                issues.append(
+                    f"{file_path.name}: 新增 NOT NULL 约束无默认值 " f"（表: {table}, 列: {col}）- 旧数据为空时迁移失败"
+                )
+            else:
+                issues.append(f"{file_path.name}: 新增 NOT NULL 约束无默认值 " "- 旧数据为空时迁移失败")
+
+    return issues
+
+
+def check_large_table_alter(file_path: Path, upgrade_content: str) -> List[str]:
+    """
+    检测大表的 ALTER TABLE 操作。
+    大表名单通过 LARGE_TABLES 配置。
+    """
+    issues = []
+
+    # 检测 alter_column 在大表上的操作
+    alter_calls = extract_alter_column_calls(upgrade_content)
+
+    for call in alter_calls:
+        table_col_match = re.search(
+            r"op\.alter_column\(\s*['\"]([^'\"]+)['\"]",
+            call,
+        )
+        if not table_col_match:
+            continue
+
+        table = table_col_match.group(1)
+        if table in LARGE_TABLES:
+            # 提取操作类型
+            ops = []
+            if re.search(r"nullable\s*=", call):
+                ops.append("修改nullable")
+            if re.search(r"type_\s*=", call):
+                ops.append("列类型变更")
+            if re.search(r"server_default\s*=", call):
+                ops.append("修改默认值")
+            if not ops:
+                ops.append("ALTER COLUMN")
+
+            issues.append(
+                f"{file_path.name}: 大表 ALTER TABLE "
+                f"（表: {table}, 操作: {', '.join(ops)}）"
+                f" - 预计行数 > {LARGE_TABLE_THRESHOLD}，可能导致长时间锁表"
+            )
+
+    # 检测大表上的 drop_column
+    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)
+        if table in LARGE_TABLES:
+            # 这个已经算高风险了，但额外加上大表提示
+            pass  # drop_column 已在 HIGH_RISK 中覆盖
+
+    return issues
+
+
 def get_new_migrations_via_diff(diff_target: str) -> List[Path]:
     """
     通过 git diff 对比目标分支/commit，找出 alembic/versions/ 下新增的迁移文件。
@@ -95,7 +222,16 @@ 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,
@@ -145,14 +281,25 @@ def analyze_migration(file_path: Path) -> Tuple[List[str], List[str], List[str]]
     medium_risks = []
     safes = []
 
+    # 高风险检测
     for pattern, desc in HIGH_RISK_PATTERNS:
         if re.search(pattern, upgrade_content):
             high_risks.append(f"{file_path.name}: {desc}")
 
+    # 中风险模式检测（通用正则）
     for pattern, desc in MEDIUM_RISK_PATTERNS:
         if re.search(pattern, upgrade_content):
             medium_risks.append(f"{file_path.name}: {desc}")
 
+    # NOT NULL 无默认值检测（专用精确检测）
+    not_null_issues = check_not_null_without_default(file_path, upgrade_content)
+    medium_risks.extend(not_null_issues)
+
+    # 大表 ALTER TABLE 检测
+    large_table_issues = check_large_table_alter(file_path, upgrade_content)
+    medium_risks.extend(large_table_issues)
+
+    # 安全操作检测
     for pattern, desc in SAFE_PATTERNS:
         if re.search(pattern, upgrade_content):
             safes.append(f"{file_path.name}: {desc}")
@@ -194,6 +341,9 @@ def main() -> int:
     if args.diff_against:
         print(f"   （对比基准：{args.diff_against}，仅检查新增迁移）")
     print()
+    print(f"📋 大表名单（ALTER TABLE 会额外告警）: {', '.join(LARGE_TABLES)}")
+    print(f"📊 大表阈值: > {LARGE_TABLE_THRESHOLD} 行")
+    print()
 
     all_high = []
     all_medium = []
@@ -247,4 +397,3 @@ def main() -> int:
 
 if __name__ == "__main__":
     sys.exit(main())
-