SQLModel 完全指南

最后更新：2026-03-29

官方文档：https://sqlmodel.tiangolo.com/
适用版本：SQLModel 0.0.21+（2026-05-07 核实）

1. 基础概念

SQLModel 是什么

SQLModel 是由 FastAPI 作者（Sebastián Ramírez）开发的库，将 SQLAlchemy（ORM）和 Pydantic（数据验证）合并为一套统一的模型定义。一个类既是数据库表，又是 API 请求/响应的 Schema。

与分开使用 SQLAlchemy + Pydantic 的对比

安装

pip install sqlmodel

# 异步支持需要异步驱动
pip install aiosqlite          # SQLite 异步
pip install asyncpg            # PostgreSQL 异步
pip install aiomysql           # MySQL 异步

SQLModel 依赖 SQLAlchemy 2.x 和 Pydantic v2，安装时会自动安装。

2. 核心模型定义

table=True vs table=False

SQLModel 模型分两类：

from sqlmodel import SQLModel, Field

# 数据库表模型（table=True）
class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(max_length=100)
    email: str = Field(unique=True, index=True)
    age: int = Field(ge=0)
    is_active: bool = Field(default=True)

# 纯 Schema 模型（table=False，默认）
class UserCreate(SQLModel):
    name: str
    email: str
    age: int

Field() 参数说明

SQLModel 的 Field() 同时支持 SQLAlchemy 列配置和 Pydantic 验证参数：

3. Schema 分层设计（与 Pydantic 配合）

SQLModel 的标准做法是用一个基类定义共有字段，再派生出不同用途的 Schema：

from sqlmodel import SQLModel, Field
from datetime import datetime

# 基础字段（所有变体共有）
class UserBase(SQLModel):
    name: str = Field(min_length=1, max_length=100, description="用户名")
    email: str = Field(description="邮箱地址")
    age: int = Field(ge=0, le=150)

# 创建时使用：不需要 id，需要密码
class UserCreate(UserBase):
    password: str = Field(min_length=8)

# 更新时使用：所有字段可选
class UserUpdate(SQLModel):
    name: str | None = Field(default=None, min_length=1, max_length=100)
    email: str | None = None
    age: int | None = Field(default=None, ge=0, le=150)

# 数据库表模型：继承基础字段 + 数据库专属字段
class User(UserBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    hashed_password: str
    is_active: bool = Field(default=True)
    created_at: datetime = Field(default_factory=datetime.now)

# API 响应：不含密码
class UserResponse(UserBase):
    id: int
    is_active: bool
    created_at: datetime

4. 数据库连接与引擎

同步引擎

from sqlmodel import create_engine, SQLModel

# SQLite（开发环境）
engine = create_engine("sqlite:///database.db", echo=True)

# PostgreSQL
engine = create_engine("postgresql+psycopg2://user:pass@localhost/dbname")

# MySQL
engine = create_engine("mysql+pymysql://user:pass@localhost/dbname")

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

if __name__ == "__main__":
    create_db_and_tables()

异步引擎

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlmodel import SQLModel

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"
# 或 SQLite 异步
# DATABASE_URL = "sqlite+aiosqlite:///database.db"

async_engine = create_async_engine(DATABASE_URL, echo=True)

AsyncSessionLocal = sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

async def create_db_and_tables():
    async with async_engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

5. CRUD 操作

同步 Session

from sqlmodel import Session, select

# 创建
def create_user(user_data: UserCreate) -> User:
    with Session(engine) as session:
        hashed_pw = hash_password(user_data.password)
        db_user = User(
            **user_data.model_dump(exclude={"password"}),
            hashed_password=hashed_pw,
        )
        session.add(db_user)
        session.commit()
        session.refresh(db_user)  # 刷新获取数据库生成的字段（如 id）
        return db_user

# 查询单条
def get_user(user_id: int) -> User | None:
    with Session(engine) as session:
        return session.get(User, user_id)

# 查询列表
def get_users(offset: int = 0, limit: int = 20) -> list[User]:
    with Session(engine) as session:
        statement = select(User).offset(offset).limit(limit)
        return session.exec(statement).all()

# 条件查询
def get_user_by_email(email: str) -> User | None:
    with Session(engine) as session:
        statement = select(User).where(User.email == email)
        return session.exec(statement).first()

# 更新
def update_user(user_id: int, user_data: UserUpdate) -> User | None:
    with Session(engine) as session:
        db_user = session.get(User, user_id)
        if not db_user:
            return None
        # exclude_unset=True 只更新传入的字段
        update_data = user_data.model_dump(exclude_unset=True)
        for key, value in update_data.items():
            setattr(db_user, key, value)
        session.add(db_user)
        session.commit()
        session.refresh(db_user)
        return db_user

# 删除
def delete_user(user_id: int) -> bool:
    with Session(engine) as session:
        db_user = session.get(User, user_id)
        if not db_user:
            return False
        session.delete(db_user)
        session.commit()
        return True

异步 Session

from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select

async def create_user(session: AsyncSession, user_data: UserCreate) -> User:
    db_user = User(
        **user_data.model_dump(exclude={"password"}),
        hashed_password=hash_password(user_data.password),
    )
    session.add(db_user)
    await session.commit()
    await session.refresh(db_user)
    return db_user

async def get_user(session: AsyncSession, user_id: int) -> User | None:
    return await session.get(User, user_id)

async def get_users(
    session: AsyncSession,
    offset: int = 0,
    limit: int = 20,
) -> list[User]:
    result = await session.exec(select(User).offset(offset).limit(limit))
    return result.all()

async def update_user(
    session: AsyncSession,
    user_id: int,
    user_data: UserUpdate,
) -> User | None:
    db_user = await session.get(User, user_id)
    if not db_user:
        return None
    for key, value in user_data.model_dump(exclude_unset=True).items():
        setattr(db_user, key, value)
    session.add(db_user)
    await session.commit()
    await session.refresh(db_user)
    return db_user

async def delete_user(session: AsyncSession, user_id: int) -> bool:
    db_user = await session.get(User, user_id)
    if not db_user:
        return False
    await session.delete(db_user)
    await session.commit()
    return True

6. 与 FastAPI 完整集成

依赖注入 Session

# src/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlmodel import SQLModel
from typing import AsyncGenerator

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"

engine = create_async_engine(DATABASE_URL, echo=True)

AsyncSessionLocal = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        yield session

async def create_db_and_tables():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

# src/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from .database import create_db_and_tables

@asynccontextmanager
async def lifespan(app: FastAPI):
    await create_db_and_tables()  # 启动时建表
    yield

app = FastAPI(lifespan=lifespan)

完整路由示例

# src/routers/users.py
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select

from ..database import get_session
from ..models import User, UserCreate, UserUpdate, UserResponse

router = APIRouter(prefix="/users", tags=["users"])

@router.post("/", response_model=UserResponse, status_code=201)
async def create_user(
    user_data: UserCreate,
    session: AsyncSession = Depends(get_session),
):
    # 检查邮箱是否已存在
    result = await session.exec(select(User).where(User.email == user_data.email))
    if result.first():
        raise HTTPException(status_code=400, detail="邮箱已注册")

    db_user = User(
        **user_data.model_dump(exclude={"password"}),
        hashed_password=hash_password(user_data.password),
    )
    session.add(db_user)
    await session.commit()
    await session.refresh(db_user)
    return db_user

@router.get("/", response_model=list[UserResponse])
async def list_users(
    offset: int = Query(default=0, ge=0),
    limit: int = Query(default=20, ge=1, le=100),
    session: AsyncSession = Depends(get_session),
):
    result = await session.exec(select(User).offset(offset).limit(limit))
    return result.all()

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,
    session: AsyncSession = Depends(get_session),
):
    db_user = await session.get(User, user_id)
    if not db_user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return db_user

@router.patch("/{user_id}", response_model=UserResponse)
async def update_user(
    user_id: int,
    user_data: UserUpdate,
    session: AsyncSession = Depends(get_session),
):
    db_user = await session.get(User, user_id)
    if not db_user:
        raise HTTPException(status_code=404, detail="用户不存在")

    for key, value in user_data.model_dump(exclude_unset=True).items():
        setattr(db_user, key, value)

    session.add(db_user)
    await session.commit()
    await session.refresh(db_user)
    return db_user

@router.delete("/{user_id}", status_code=204)
async def delete_user(
    user_id: int,
    session: AsyncSession = Depends(get_session),
):
    db_user = await session.get(User, user_id)
    if not db_user:
        raise HTTPException(status_code=404, detail="用户不存在")
    await session.delete(db_user)
    await session.commit()

7. 关联关系

一对多关系

from typing import Optional
from sqlmodel import SQLModel, Field, Relationship

class Team(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(unique=True)

    # 反向关系（一个 Team 有多个 User）
    users: list["User"] = Relationship(back_populates="team")

class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    team_id: int | None = Field(default=None, foreign_key="team.id")

    # 正向关系（一个 User 属于一个 Team）
    team: Optional[Team] = Relationship(back_populates="users")

多对多关系（通过关联表）

from sqlmodel import SQLModel, Field, Relationship

# 关联表模型
class ArticleTagLink(SQLModel, table=True):
    article_id: int | None = Field(
        default=None, foreign_key="article.id", primary_key=True
    )
    tag_id: int | None = Field(
        default=None, foreign_key="tag.id", primary_key=True
    )

class Article(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str
    tags: list["Tag"] = Relationship(
        back_populates="articles", link_model=ArticleTagLink
    )

class Tag(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(unique=True)
    articles: list[Article] = Relationship(
        back_populates="tags", link_model=ArticleTagLink
    )

关联查询（避免 N+1）

from sqlalchemy.orm import selectinload
from sqlmodel import select

# 使用 selectinload 预加载关联数据，避免 N+1 查询
async def get_teams_with_users(session: AsyncSession) -> list[Team]:
    result = await session.exec(
        select(Team).options(selectinload(Team.users))
    )
    return result.all()

8. 查询进阶

条件查询与排序

from sqlmodel import select, col

# 多条件 AND
statement = select(User).where(
    User.is_active == True,
    User.age >= 18,
)

# OR 条件
from sqlalchemy import or_
statement = select(User).where(
    or_(User.name == "Alice", User.email == "[email protected]")
)

# 模糊查询
statement = select(User).where(User.name.contains("alice"))
statement = select(User).where(User.email.startswith("alice"))

# 排序
statement = select(User).order_by(User.created_at.desc())
statement = select(User).order_by(col(User.name).asc())

# 统计
from sqlalchemy import func
result = await session.exec(select(func.count(User.id)))
count = result.one()

分页查询

async def paginate_users(
    session: AsyncSession,
    page: int = 1,
    page_size: int = 20,
) -> tuple[list[User], int]:
    offset = (page - 1) * page_size

    # 数据
    result = await session.exec(
        select(User).offset(offset).limit(page_size)
    )
    users = result.all()

    # 总数
    count_result = await session.exec(select(func.count(User.id)))
    total = count_result.one()

    return users, total

原生 SQL（复杂查询）

from sqlalchemy import text

async def search_users(session: AsyncSession, keyword: str) -> list[User]:
    result = await session.exec(
        text("SELECT * FROM user WHERE name LIKE :keyword"),
        {"keyword": f"%{keyword}%"},
    )
    return result.all()

9. 数据验证与序列化（Pydantic 层）

SQLModel 模型完全兼容 Pydantic v2 的验证器和序列化特性。

在 SQLModel 中使用 Pydantic 验证器

from sqlmodel import SQLModel, Field
from pydantic import field_validator, model_validator
from typing import Self

class UserCreate(SQLModel):
    name: str = Field(min_length=1, max_length=100)
    email: str
    password: str = Field(min_length=8)
    confirm_password: str

    @field_validator("email")
    @classmethod
    def email_to_lower(cls, v: str) -> str:
        return v.lower().strip()

    @model_validator(mode="after")
    def passwords_match(self) -> Self:
        if self.password != self.confirm_password:
            raise ValueError("两次密码不一致")
        return self

model_dump / model_validate

# 从字典创建
user = UserCreate.model_validate({
    "name": "Alice",
    "email": "[email protected]",
    "password": "secret123",
    "confirm_password": "secret123",
})

# PATCH 更新：只取传入的字段
user_update = UserUpdate(name="Bob")
update_fields = user_update.model_dump(exclude_unset=True)
# {'name': 'Bob'}

# 从数据库对象转为响应 Schema
db_user: User = ...
response = UserResponse.model_validate(db_user)

从数据库模型到响应 Schema 的转换

SQLModel 的 table 模型可以直接作为 Pydantic 模型使用，FastAPI 的 response_model 会自动做字段过滤：

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, session: AsyncSession = Depends(get_session)):
    db_user = await session.get(User, user_id)
    if not db_user:
        raise HTTPException(404, "用户不存在")
    # 直接返回 ORM 对象，FastAPI 通过 response_model 自动序列化和过滤字段
    return db_user

10. Alembic 数据库迁移

SQLModel 与 Alembic 配合进行数据库迁移管理。

初始化

pip install alembic
alembic init alembic

配置 alembic/env.py

# alembic/env.py
from sqlmodel import SQLModel
from src.models import *  # 导入所有模型，确保它们被注册到 metadata

target_metadata = SQLModel.metadata

常用命令

# 自动检测模型变更，生成迁移脚本
alembic revision --autogenerate -m "add user table"

# 执行迁移（升级到最新版本）
alembic upgrade head

# 回滚一个版本
alembic downgrade -1

# 查看迁移历史
alembic history

# 查看当前版本
alembic current

11. 完整项目结构

src/
  __init__.py
  main.py              # FastAPI app 入口，lifespan 建表
  database.py          # 引擎、Session 依赖
  models/
    __init__.py        # 统一导出所有模型
    user.py            # User 相关 SQLModel 模型
    article.py
  routers/
    __init__.py
    users.py
    articles.py
  services/            # 业务逻辑层（可选）
    user_service.py
  core/
    config.py          # pydantic-settings 配置
    security.py        # 密码哈希、JWT
alembic/
  versions/
  env.py
alembic.ini

# src/models/__init__.py
from .user import User, UserBase, UserCreate, UserUpdate, UserResponse
from .article import Article, ArticleCreate, ArticleResponse

# src/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from .database import create_db_and_tables
from .routers import users, articles

@asynccontextmanager
async def lifespan(app: FastAPI):
    await create_db_and_tables()
    yield

app = FastAPI(title="My API", lifespan=lifespan)
app.include_router(users.router)
app.include_router(articles.router)

12. 常用代码段

通用分页响应体

from typing import Generic, TypeVar
from sqlmodel import SQLModel

T = TypeVar("T")

class Page(SQLModel, Generic[T]):
    items: list[T]
    total: int
    page: int
    page_size: int
    total_pages: int

    @classmethod
    def create(cls, items: list[T], total: int, page: int, page_size: int) -> "Page[T]":
        return cls(
            items=items,
            total=total,
            page=page,
            page_size=page_size,
            total_pages=(total + page_size - 1) // page_size,
        )

带软删除的基类

from datetime import datetime
from sqlmodel import SQLModel, Field

class SoftDeleteMixin(SQLModel):
    deleted_at: datetime | None = Field(default=None)

    @property
    def is_deleted(self) -> bool:
        return self.deleted_at is not None

class Article(SoftDeleteMixin, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str

# 查询时过滤已删除记录
statement = select(Article).where(Article.deleted_at == None)

带时间戳的基类

from datetime import datetime
from sqlmodel import SQLModel, Field

class TimestampMixin(SQLModel):
    created_at: datetime = Field(default_factory=datetime.now)
    updated_at: datetime = Field(default_factory=datetime.now)

class User(TimestampMixin, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str

通用 CRUD 基类

from typing import Generic, TypeVar, Type
from sqlmodel import SQLModel, select
from sqlalchemy.ext.asyncio import AsyncSession

ModelType = TypeVar("ModelType", bound=SQLModel)
CreateSchemaType = TypeVar("CreateSchemaType", bound=SQLModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=SQLModel)

class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
    def __init__(self, model: Type[ModelType]):
        self.model = model

    async def get(self, session: AsyncSession, id: int) -> ModelType | None:
        return await session.get(self.model, id)

    async def get_multi(
        self, session: AsyncSession, *, offset: int = 0, limit: int = 20
    ) -> list[ModelType]:
        result = await session.exec(
            select(self.model).offset(offset).limit(limit)
        )
        return result.all()

    async def create(
        self, session: AsyncSession, obj_in: CreateSchemaType
    ) -> ModelType:
        db_obj = self.model.model_validate(obj_in)
        session.add(db_obj)
        await session.commit()
        await session.refresh(db_obj)
        return db_obj

    async def update(
        self, session: AsyncSession, db_obj: ModelType, obj_in: UpdateSchemaType
    ) -> ModelType:
        for key, value in obj_in.model_dump(exclude_unset=True).items():
            setattr(db_obj, key, value)
        session.add(db_obj)
        await session.commit()
        await session.refresh(db_obj)
        return db_obj

    async def delete(self, session: AsyncSession, id: int) -> bool:
        obj = await session.get(self.model, id)
        if not obj:
            return False
        await session.delete(obj)
        await session.commit()
        return True

# 使用
user_crud = CRUDBase[User, UserCreate, UserUpdate](User)

13. 最佳实践

模型继承层次

UserBase（纯 SQLModel，共有字段 + Pydantic 验证）
  ├── UserCreate（+ password）
  ├── UserUpdate（所有字段可选）
  ├── UserResponse（+ id、created_at，无敏感字段）
  └── User（table=True，+ hashed_password、ORM 专属字段）

response_model 过滤敏感字段

利用 FastAPI 的 response_model 参数自动过滤，不要手动序列化：

# 返回 User ORM 对象，FastAPI 自动按 UserResponse 过滤字段
@router.post("/", response_model=UserResponse)
async def create_user(data: UserCreate, session=Depends(get_session)):
    ...
    return db_user  # 直接返回，框架做过滤

使用 exclude_unset=True 实现 PATCH 语义

@router.patch("/{user_id}", response_model=UserResponse)
async def update_user(user_id: int, data: UserUpdate, session=Depends(get_session)):
    db_user = await session.get(User, user_id)
    # 只更新客户端实际传入的字段
    update_data = data.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_user, key, value)
    ...

Session 不要跨请求共享

每个请求应该独立使用一个 Session，通过 Depends(get_session) 注入，不要在全局或多个请求间共享同一个 Session 实例。

不要在 SQLModel 模型里直接写业务逻辑

保持 models/ 只负责定义结构，业务逻辑放在 services/ 层，路由层只做参数接收和调用：

router（接收请求、注入依赖）
  → service（业务逻辑、事务控制）
    → crud / session（数据库操作）

14. 踩坑与注意事项

table=True 的模型不能作为纯 Pydantic 模型实例化时传入其他 table=True 字段

带 table=True 的模型在定义关联关系时，关联对象会触发 SQLAlchemy lazy load，在异步上下文中会报错：

# 关联字段要么在查询时用 selectinload 预加载，要么使用 noload
from sqlalchemy.orm import selectinload, noload

# 预加载（需要用到关联数据时）
statement = select(User).options(selectinload(User.team))

# 禁用加载（不需要关联数据时，避免 lazy load 报错）
statement = select(User).options(noload(User.team))

异步 Session 必须用 await session.exec()

# 错误：同步写法在异步 session 里不工作
result = session.exec(select(User))

# 正确
result = await session.exec(select(User))

model_validate 与直接实例化的区别

# 直接实例化：跳过部分验证逻辑
user = User(name="Alice", email="[email protected]")

# model_validate：走完整 Pydantic 验证流程（推荐）
user = User.model_validate({"name": "Alice", "email": "[email protected]"})

# 从另一个 Pydantic/SQLModel 对象创建
db_user = User.model_validate(user_create)

SQLite 不支持并发写入

开发用 SQLite 时注意它不支持多并发写入。生产环境请使用 PostgreSQL 或 MySQL。若必须在开发时测试并发，SQLite 的 WAL 模式可以缓解：

from sqlalchemy import event

engine = create_async_engine("sqlite+aiosqlite:///database.db")

@event.listens_for(engine.sync_engine, "connect")
def set_sqlite_pragma(dbapi_conn, _):
    cursor = dbapi_conn.cursor()
    cursor.execute("PRAGMA journal_mode=WAL")
    cursor.close()

字段默认值与数据库默认值的区别

class User(SQLModel, table=True):
    # Python 层默认值：每次创建对象时由 Python 赋值
    created_at: datetime = Field(default_factory=datetime.now)

    # 数据库层默认值：由数据库 server_default 提供（Python 对象上先为 None）
    updated_at: datetime | None = Field(
        default=None,
        sa_column_kwargs={"server_default": "CURRENT_TIMESTAMP"},
    )

最佳实践

table=True 模型与纯 Pydantic 模型分开定义：同一个类同时用于 ORM 映射和 API schema 会带来约束冲突。推荐模式：基础字段定义在无 table 的基类，table=True 的 ORM 类继承它，API schema 单独定义：

class UserBase(SQLModel):
    name: str
    email: str

class User(UserBase, table=True):
    id: int | None = Field(default=None, primary_key=True)

class UserCreate(UserBase):
    password: str  # 仅写入时使用，不在 ORM 模型中

关系字段用 Relationship 而非原始外键：直接操作外键 ID 容易出现 N+1 查询，用 Relationship + selectin 加载策略批量预加载关联对象：

from sqlmodel import Relationship

class Team(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    members: list["User"] = Relationship(back_populates="team",
                                          sa_relationship_kwargs={"lazy": "selectin"})

异步场景用 AsyncSession，不混用同步 Session：SQLModel 的异步支持基于 SQLAlchemy async，create_async_engine 配 AsyncSession，勿在同一事务中混用同步/异步操作。

迁移用 Alembic，不用 create_all 做生产更新：SQLModel.metadata.create_all() 只新建表，不做 ALTER，生产环境必须用 Alembic 管理 schema 变更，保留升降级能力。

Field(index=True) 为高频查询字段加索引：email、created_at 等 WHERE 条件字段忘加索引会导致全表扫描，在 Field() 中声明 index=True 或 unique=True 会生成对应索引。

常见陷阱

陷阱：table=True 类在同一进程中被重复定义

现象： 测试时报 Table 'xxx' is already defined for this MetaData instance。
原因： 多个测试模块 import 了同一 SQLModel 类，SQLAlchemy MetaData 禁止重复注册同名表。
解决： 所有 table=True 类在同一个模块中定义并集中 import，测试中避免多次 create_all；或使用 extend_existing=True（不推荐生产使用）。

陷阱：Optional 字段在数据库中不允许 NULL

现象： 模型字段标注 name: str | None = None，但 INSERT 时报 NOT NULL constraint failed。
原因： SQLModel 的 nullable 推断有时与 Python 类型注解不完全一致，需要显式声明。
解决： 对可空字段显式设置 Field(nullable=True)：

name: str | None = Field(default=None, nullable=True)

陷阱：关系加载时 Session 已关闭导致 DetachedInstanceError

现象： 在 with Session(engine) as session: 外访问 user.team 时报 DetachedInstanceError。
原因： Session 关闭后懒加载（lazy load）无法建立新查询，尝试访问关系属性失败。
解决： 在 Session 内完成所有关系访问，或改用 selectin / joined 加载策略在查询时预加载。

参见

FastAPI完全指南
SQLAlchemy完全指南
Pydantic完全指南