> ## Content Index
> Fetch the complete content index at: https://blog.vercanti.com/llms.txt
> Use this file to discover other available public pages before exploring further.

# SQLModel 完全指南
- URL: https://blog.vercanti.com/sqlmodel-wan-quan-zhi-nan/
- Published: 2026-08-28T14:34:45.000Z
- Updated: 2026-08-28T14:57:15.000Z
- Description: 最后更新：2026-03-29 SQLModel 是由 FastAPI 作者（Sebastián Ramírez）开发的库，将 SQLAlchemy（ORM）和 Pydantic（数据验证）合并为一套统一的模型定义。一个类既是数据库表，又是 API 请求/响应的 Schema。 SQLModel 依赖 SQLAlchemy 2.x 和 Pydantic v2，安装时会自动安装。 SQLModel 模型分两类： SQLModel 的 Field() 同时支持 SQLAlchemy 列配置和 Pydantic 验证参数： SQLModel 的标准做法是用一个
- Author: yellowdog
- Tags: Python, 框架与库

最后更新：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 | 数据库 ORM、连接管理、SQL 执行 |
| Pydantic   | 数据验证、序列化、类型系统       |
| SQLModel   | 两者的融合层，一个模型同时具备两者能力 |

### 与分开使用 SQLAlchemy + Pydantic 的对比

| 场景   | 分开使用                              | SQLModel  |
| ---- | --------------------------------- | --------- |
| 模型数量 | 每张表需要 ORM 模型 + 多个 Pydantic Schema | 一个基类，多个变体 |
| 字段同步 | 字段修改需要同步两处                        | 只改一处      |
| 类型支持 | 分别类型检查                            | 统一类型推断    |
| 学习成本 | 需要掌握两套 API                        | 统一 API    |
| 灵活性  | 高（各自独立可深度定制）                      | 略低（有约束）   |

### 安装

```bash
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 模型分两类：

| 参数              | 类型            | 说明                    |
| --------------- | ------------- | --------------------- |
| table=True      | 数据库表模型        | 对应数据库中的一张表，可用于 ORM 操作 |
| table=False（默认） | 纯 Pydantic 模型 | 只用于数据验证/序列化，不映射数据库    |

```python
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 验证参数：

| 参数                          | 说明                                |
| --------------------------- | --------------------------------- |
| primary\_key=True           | 设为主键                              |
| default=None                | 默认值（主键通常设为 None，由数据库自增）           |
| default\_factory            | 动态默认值工厂                           |
| index=True                  | 创建数据库索引                           |
| unique=True                 | 唯一约束                              |
| nullable=False              | 不可为 NULL（默认根据类型注解推断）              |
| foreign\_key="table.column" | 外键引用                              |
| max\_length                 | 字符串最大长度（同时影响数据库列类型和 Pydantic 验证）  |
| ge / le / gt / lt           | 数值范围验证（仅 Pydantic 层，不影响数据库）       |
| sa\_column                  | 直接传入 SQLAlchemy Column 对象（用于高级配置） |
| sa\_column\_kwargs          | 传给 SQLAlchemy Column 的额外参数        |
| description                 | 字段描述（出现在 JSON Schema / API 文档中）   |
| title                       | 字段标题                              |
| alias                       | 字段别名                              |

---

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

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

```python
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\. 数据库连接与引擎

### 同步引擎

```python
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()

```

### 异步引擎

```python
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

```python
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

```python
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

```python
# 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)

```

```python
# 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)

```

### 完整路由示例

```python
# 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\. 关联关系

### 一对多关系

```python
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")

```

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

```python
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）

```python
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\. 查询进阶

### 条件查询与排序

```python
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 == "alice@example.com")
)

# 模糊查询
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()

```

### 分页查询

```python
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（复杂查询）

```python
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 验证器

```python
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

```python
# 从字典创建
user = UserCreate.model_validate({
    "name": "Alice",
    "email": "Alice@Example.com",
    "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` 会自动做字段过滤：

```python
@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 配合进行数据库迁移管理。

### 初始化

```bash
pip install alembic
alembic init alembic

```

### 配置 alembic/env.py

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

target_metadata = SQLModel.metadata

```

### 常用命令

```bash
# 自动检测模型变更，生成迁移脚本
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

```

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

```

```python
# 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\. 常用代码段

### 通用分页响应体

```python
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,
        )

```

### 带软删除的基类

```python
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)

```

### 带时间戳的基类

```python
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 基类

```python
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` 参数自动过滤，不要手动序列化：

```python
# 返回 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 语义

```python
@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，在异步上下文中会报错：

```python
# 关联字段要么在查询时用 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()

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

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

```

### model\_validate 与直接实例化的区别

```python
# 直接实例化：跳过部分验证逻辑
user = User(name="Alice", email="alice@example.com")

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

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

```

### SQLite 不支持并发写入

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

```python
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()

```

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

```python
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 单独定义：

```python
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` 加载策略批量预加载关联对象：

```python
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)`：

```python
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完全指南](https://blog.vercanti.com/fastapi-wan-quan-zhi-nan/)  
[SQLAlchemy完全指南](https://blog.vercanti.com/sqlalchemy-wan-quan-zhi-nan/)  
[Pydantic完全指南](https://blog.vercanti.com/pydantic-wan-quan-zhi-nan/)