ci: require TypeBase for new models (#44087)

Co-authored-by: Codex <codex@openai.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

.github/workflows/main-ci.yml

@@ -64,6 +64,8 @@ jobs:
               - 'scripts/ast_grep_guard.py'
               - 'scripts/check_no_new_getattr.py'
               - 'scripts/check_no_spec_mock.py'
+              - 'scripts/check_no_new_base_model.py'
+              - 'scripts/ast_grep_rules/no_new_base_model.yml'
               - 'scripts/check_no_new_test_doubles.py'
               - 'scripts/check_no_new_controller_sqlalchemy.py'
               - 'scripts/lint_controller_sqlalchemy.py'
@@ -167,6 +169,8 @@ jobs:
               - 'scripts/ast_grep_guard.py'
               - 'scripts/check_no_new_getattr.py'
               - 'scripts/check_no_spec_mock.py'
+              - 'scripts/check_no_new_base_model.py'
+              - 'scripts/ast_grep_rules/no_new_base_model.yml'
               - 'scripts/check_no_new_test_doubles.py'
               - 'scripts/check_no_new_controller_sqlalchemy.py'
               - 'scripts/check_no_new_session_mock.py'

.github/workflows/style.yml

@@ -97,6 +97,12 @@ jobs:
         if: inputs.run-python-style
         run: uv run --project api python scripts/check_no_spec_mock.py
 
+      - name: Run No New Base Model Guard
+        if: inputs.run-python-style
+        env:
+          GUARD_BASE_REV: ${{ github.event_name == 'pull_request' && 'HEAD^1' || inputs.base-rev }}
+        run: uv run --project api python scripts/check_no_new_base_model.py --base-rev "$GUARD_BASE_REV"
+
       - name: Run No New Test Doubles Guard
         if: inputs.run-python-style || inputs.run-dify-agent-style
         # A PR event can retain an older base SHA after its target branch advances.

Makefile

@@ -81,6 +81,7 @@ lint:
 	@uv run --project api --dev ruff check --fix ./api
 	@$(MAKE) api-contract-lint
 	@uv run --project api python scripts/check_no_spec_mock.py
+	@uv run --project api python scripts/check_no_new_base_model.py
 	@uv run --directory api --dev lint-imports
 	@uv run --project api --dev dotenv-linter ./api/.env.example ./web/.env.example
 	@echo "✅ Linting complete"

api/AGENTS.md

@@ -25,6 +25,11 @@ Run direct Python commands through `uv run --project api`. Docker-backed integra
 - Use existing Celery task and queue owners for asynchronous work; do not route unrelated jobs through workflow-specific services.
 - Celery tasks that may be retried or redelivered must keep side effects idempotent and log affected resource identifiers.
 
+New SQLAlchemy models must inherit from `models.base.TypeBase`. The ast-grep
+guard rejects new `Base` subclasses. Existing `Base` subclasses may be edited
+or migrated without adding more legacy inheritance. `make lint` checks local
+changes, and CI checks the PR against its target revision.
+
 ## Typing Best Practises
 
 - Treat nullable types (`T | None`) strictly as union types; document the semantic meaning for each branch.

api/tests/unit_tests/commands/test_check_no_new_base_model.py

@@ -0,0 +1,137 @@
+"""Exercise the TypeBase guard with real Git revisions and ast-grep scans."""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+SCRIPT = Path(__file__).resolve().parents[4] / "scripts" / "check_no_new_base_model.py"
+SOURCE_PATH = "api/models/example.py"
+
+
+def git(repo: Path, *args: str) -> str:
+    return subprocess.run(["git", *args], cwd=repo, text=True, capture_output=True, check=True).stdout.strip()
+
+
+def write_source(repo: Path, source: str, path: str = SOURCE_PATH) -> None:
+    target = repo / path
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text(source)
+
+
+def commit(repo: Path) -> str:
+    git(repo, "add", ".")
+    git(repo, "commit", "-m", "guard fixture")
+    return git(repo, "rev-parse", "HEAD")
+
+
+@pytest.fixture
+def repo(tmp_path: Path) -> Path:
+    git(tmp_path, "init", "-b", "main")
+    git(tmp_path, "config", "user.name", "Guard Tests")
+    git(tmp_path, "config", "user.email", "guard@example.com")
+    write_source(tmp_path, "value = 1\n")
+    commit(tmp_path)
+    return tmp_path
+
+
+def run_guard(repo: Path, *args: str) -> subprocess.CompletedProcess[str]:
+    return subprocess.run([sys.executable, str(SCRIPT), *args], cwd=repo, text=True, capture_output=True, check=False)
+
+
+@pytest.mark.parametrize(
+    ("imports", "base"),
+    [
+        ("from models.base import Base", "Base"),
+        ("from .base import Base", "Base"),
+        ("from ..base import Base", "Base"),
+        ("from api.models.base import Base", "Base"),
+        ("from models.base import Base as Legacy", "Legacy"),
+        ("from .base import Base as Legacy", "Legacy"),
+        ("import models.base as model_base", "model_base.Base"),
+        ("from models import base as model_base", "model_base.Base"),
+        ("from models import base", "base.Base"),
+        ("import models.base", "models.base.Base"),
+        ("import api.models.base", "api.models.base.Base"),
+    ],
+)
+@pytest.mark.parametrize("multiline", [False, True])
+def test_rejects_new_legacy_model(repo: Path, imports: str, base: str, multiline: bool) -> None:
+    header = f"class NewModel(\n    Mixin,\n    {base},\n):" if multiline else f"class NewModel(Mixin, {base}):"
+    write_source(repo, f"{imports}\n{header}\n    pass\n")
+    git(repo, "add", SOURCE_PATH)
+    result = run_guard(repo, "--staged")
+    assert result.returncode == 1, result.stderr
+    assert "NewModel must inherit from TypeBase" in result.stderr
+
+
+@pytest.mark.parametrize(
+    "source",
+    [
+        "from models.base import TypeBase\nclass NewModel(TypeBase): pass\n",
+        "from models.base import TypeBase as Base\nclass NewModel(Base): pass\n",
+        "from external import Base\nclass NewModel(Base): pass\n",
+        "from external import Base as Legacy\nclass NewModel(Legacy): pass\n",
+        "from external import base as model_base\nclass NewModel(model_base.Base): pass\n",
+        "# class NewModel(Base): pass\nvalue = 'class NewModel(Base): pass'\n",
+        "from models.base import Base\nclass NewModel:\n    value = Base\n",
+    ],
+)
+def test_allows_typed_models_and_unrelated_syntax(repo: Path, source: str) -> None:
+    write_source(repo, source)
+    git(repo, "add", SOURCE_PATH)
+    result = run_guard(repo, "--staged")
+    assert result.returncode == 0, result.stderr
+
+
+def test_existing_legacy_model_can_be_edited(repo: Path) -> None:
+    write_source(repo, "from models.base import Base\nclass Existing(Base):\n    value = 1\n")
+    base = commit(repo)
+    write_source(repo, "from models.base import Base\nclass Existing(Base):\n    value = 2\n")
+    commit(repo)
+    result = run_guard(repo, "--base-rev", base)
+    assert result.returncode == 0, result.stderr
+
+
+def test_replacing_one_legacy_model_with_another_is_rejected(repo: Path) -> None:
+    write_source(repo, "from models.base import Base\nclass Existing(Base): pass\n")
+    base = commit(repo)
+    write_source(repo, "from models.base import Base\nclass NewModel(Base): pass\n")
+    commit(repo)
+    result = run_guard(repo, "--base-rev", base)
+    assert result.returncode == 1, result.stderr
+
+
+def test_import_change_can_introduce_forbidden_inheritance(repo: Path) -> None:
+    write_source(repo, "from external import Base\nclass Existing(Base): pass\n")
+    base = commit(repo)
+    write_source(repo, "from models.base import Base\nclass Existing(Base): pass\n")
+    commit(repo)
+    result = run_guard(repo, "--base-rev", base)
+    assert result.returncode == 1, result.stderr
+
+
+def test_git_rename_keeps_existing_model_baseline(repo: Path) -> None:
+    write_source(repo, "from models.base import Base\nclass Existing(Base): pass\n")
+    base = commit(repo)
+    git(repo, "mv", SOURCE_PATH, "api/models/renamed.py")
+    commit(repo)
+    result = run_guard(repo, "--base-rev", base)
+    assert result.returncode == 0, result.stderr
+
+
+def test_local_check_includes_unstaged_and_untracked_models(repo: Path) -> None:
+    write_source(repo, "from models.base import Base\nclass Unstaged(Base): pass\n")
+    write_source(repo, "from models.base import Base\nclass Untracked(Base): pass\n", "api/models/untracked.py")
+    result = run_guard(repo)
+    assert result.returncode == 1, result.stderr
+    assert "Unstaged" in result.stderr
+    assert "Untracked" in result.stderr
+
+
+def test_invalid_baseline_fails(repo: Path) -> None:
+    result = run_guard(repo, "--base-rev", "missing-revision")
+    assert result.returncode == 2

scripts/ast_grep_rules/no_new_base_model.yml

@@ -0,0 +1,93 @@
+id: no-new-base-model
+language: python
+severity: error
+message: New SQLAlchemy models must inherit from TypeBase instead of Base.
+utils:
+  project-import:
+    kind: import_from_statement
+    has:
+      field: module_name
+      regex: '^(models\.base|api\.models\.base|\.+base)$'
+  project-base:
+    any:
+      - all:
+          - pattern: Base
+          - inside:
+              kind: module
+              stopBy: end
+              has:
+                stopBy: end
+                all:
+                  - matches: project-import
+                  - has:
+                      stopBy: neighbor
+                      all:
+                        - kind: dotted_name
+                        - regex: '^Base$'
+      - regex: '^(api\.)?models\.base\.Base$'
+      - all:
+          - pattern: $BASE
+          - inside:
+              kind: module
+              stopBy: end
+              has:
+                stopBy: end
+                all:
+                  - pattern:
+                      context: from models.base import Base as $BASE
+                      selector: aliased_import
+                  - inside:
+                      stopBy: neighbor
+                      matches: project-import
+      - all:
+          - pattern: $MODULE.Base
+          - inside:
+              kind: module
+              stopBy: end
+              has:
+                stopBy: end
+                any:
+                  - all:
+                      - pattern:
+                          context: import models.base as $MODULE
+                          selector: aliased_import
+                      - has:
+                          field: name
+                          regex: '^(api\.)?models\.base$'
+                  - all:
+                      - pattern:
+                          context: from models import base as $MODULE
+                          selector: aliased_import
+                      - inside:
+                          kind: import_from_statement
+                          stopBy: neighbor
+                          has:
+                            field: module_name
+                            regex: '^(api\.)?models$'
+      - all:
+          - pattern: base.Base
+          - inside:
+              kind: module
+              stopBy: end
+              has:
+                stopBy: end
+                all:
+                  - kind: import_from_statement
+                  - has:
+                      field: module_name
+                      regex: '^(api\.)?models$'
+                  - has:
+                      stopBy: neighbor
+                      all:
+                        - kind: dotted_name
+                        - regex: '^base$'
+rule:
+  all:
+    - kind: class_definition
+    - has:
+        field: name
+        pattern: $NAME
+    - has:
+        field: superclasses
+        has:
+          matches: project-base

scripts/check_no_new_base_model.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+"""Require TypeBase for new SQLAlchemy model inheritance.
+
+Existing Base subclasses may be edited or migrated. New classes that inherit
+from models.base.Base fail, including import aliases and qualified bases.
+Class names preserve the baseline even when surrounding lines change. Renames
+preserve the baseline through Git's rename detection.
+
+CI compares committed revisions with --base-rev. Local checks inspect the
+working tree and untracked Python files against the merge base with main.
+Use --staged to inspect only staged changes against HEAD.
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from collections import Counter
+from pathlib import Path
+
+from ast_grep_guard import (
+    Violation,
+    git_output,
+    print_violations,
+    rule_path,
+    run_ast_grep,
+)
+from check_no_new_test_doubles import changed_file_versions
+
+
+def working_file_versions() -> dict[str, tuple[str, str]]:
+    baseline = "HEAD"
+    for revision in ("upstream/main", "origin/main", "main"):
+        merge_base = git_output("merge-base", "HEAD", revision, allow_missing=True).strip()
+        if merge_base:
+            baseline = merge_base
+            break
+    fields = iter(
+        git_output(
+            "diff",
+            "--name-status",
+            "-z",
+            "--find-renames",
+            "--diff-filter=AMR",
+            baseline,
+            "--",
+            "api",
+        )
+        .rstrip("\0")
+        .split("\0")
+    )
+    paths: dict[str, str] = {}
+    for status in fields:
+        if not status:
+            continue
+        old_path = next(fields)
+        path = next(fields) if status.startswith("R") else old_path
+        paths[path] = old_path
+    for path in git_output("ls-files", "--others", "--exclude-standard", "-z", "api").rstrip("\0").split("\0"):
+        paths[path] = path
+    versions = {}
+    for path in sorted(paths):
+        source = Path(path)
+        if source.suffix not in {".py", ".pyi"} or not source.is_file():
+            continue
+        versions[path] = (
+            git_output("show", f"{baseline}:{paths[path]}", allow_missing=True),
+            source.read_text(),
+        )
+    return versions
+
+
+def find_new_base_models(versions: dict[str, tuple[str, str]]) -> list[Violation]:
+    rule = rule_path("no_new_base_model.yml")
+    violations = []
+    for path, (old_source, new_source) in versions.items():
+        if not Path(path).is_relative_to("api"):
+            continue
+        existing = Counter(match.meta_variables["NAME"] for match in run_ast_grep(old_source, rule=rule))
+        for match in run_ast_grep(new_source, rule=rule):
+            name = match.meta_variables["NAME"]
+            if existing[name]:
+                existing[name] -= 1
+            else:
+                violations.append(
+                    Violation(
+                        path,
+                        match.line_number,
+                        f"no-new-base-model: {name} must inherit from TypeBase.",
+                    )
+                )
+    return violations
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    source = parser.add_mutually_exclusive_group()
+    source.add_argument("--base-rev", help="Compare this revision with committed HEAD.")
+    source.add_argument("--staged", action="store_true", help="Compare staged changes with HEAD.")
+    args = parser.parse_args()
+    try:
+        versions = changed_file_versions(args) if args.base_rev or args.staged else working_file_versions()
+        violations = find_new_base_models(versions)
+    except (OSError, RuntimeError) as error:
+        sys.stderr.write(f"{error}\n")
+        return 2
+    print_violations(violations)
+    return int(bool(violations))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())