FastAPI+SQLAlchemy写入PostgreSQL JSONB字段异常排查

问题场景 #

使用FastAPI、SQLAlchemy ORM及Pydantic操作PostgreSQL时，JSONB字段state_transition_data出现两类异常：

• 当Pydantic模型字段注解为JSON | None时，JSONB字段无法更新，推测是传入数据含单引号导致报错；
• 改用Any | None注解时，字段能更新但数据被转义，后续无法执行JSON查询。

相关代码 #

SQLAlchemy模型

from sqlalchemy import UUID, JSONB
from sqlalchemy.orm import Mapped, mapped_column
import uuid
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Entry_A(Base):
    __tablename__ = "entry_a"
    id: Mapped[UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, index=True, default=uuid.uuid4)
    state_transition_data: Mapped[JSONB] = mapped_column(JSONB, nullable=True)

Pydantic模型（原异常版本）

from pydantic import BaseModel, Json
from typing import Any, UUID

class Entry_A(BaseModel):
    id: UUID
    state_transition_data: Json | None  # 异常1：无法更新
    # 或改为 state_transition_data: Any | None  # 异常2：数据转义

解决方案 #

1. 修正Pydantic字段注解 #

使用Dict[str, Any] | None替代Json或Any，明确字段为字典类型，既匹配JSON结构，又能让Pydantic正确完成序列化/反序列化：

from typing import Dict, Any, UUID
from pydantic import BaseModel

class Entry_A(BaseModel):
    id: UUID
    state_transition_data: Dict[str, Any] | None = None

2. 确保类型映射一致性 #

SQLAlchemy的JSONB字段原生对应Python字典类型，无需额外转义。FastAPI接口接收请求时，直接用上述Pydantic模型作为请求体，框架会自动将JSON请求解析为字典。

3. 优化CRUD操作的数据处理 #

更新时直接传入Pydantic模型解析后的字典，避免手动序列化JSON字符串：

from sqlalchemy.orm import Session

def update_entry_a(db: Session, entry_id: UUID, update_data: Entry_A):
    db_entry = db.query(Entry_A).filter(Entry_A.id == entry_id).first()
    if db_entry:
        # 仅更新传入的非空字段
        update_dict = update_data.model_dump(exclude_unset=True)
        for key, value in update_dict.items():
            setattr(db_entry, key, value)
        db.commit()
        db.refresh(db_entry)
    return db_entry

4. 验证JSON查询功能 #

更新完成后，可直接使用PostgreSQL的JSONB查询语法，例如：

# 查询state_transition_data中status字段为completed的记录
results = db.query(Entry_A).filter(
    Entry_A.state_transition_data["status"].astext == "completed"
).all()

内容的提问来源于stack exchange，提问作者Phil