如何用SQLAlchemy ORM实现JSON列嵌套字段的部分更新？

你的需求完全可行，核心思路是避免直接修改内存中的data字典，转而记录待更新的JSON字段，在Session flush阶段通过数据库的json_set函数执行原子更新，这样既保留了user.field_a = 5555的直观语法，又不会覆盖JSON列中的其他字段。

具体实现步骤 #

1. 修改User模型，添加待更新字段记录逻辑

在模型中新增一个内部属性用于记录待更新的JSON路径和值，同时调整field_a的setter逻辑：

from sqlalchemy import Column, Integer, JSON, update, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import Session, event

Base = declarative_base()

class User(Base):
    __tablename__ = "user_account"

    id = Column(Integer, primary_key=True)
    data = Column(JSON, default={})

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # 用于记录待更新的JSON字段（路径: 值）
        self._pending_json_updates = {}

    @hybrid_property
    def field_a(self):
        # 优先返回待更新的值，未提交时保持一致性
        return self._pending_json_updates.get('$.field_a', self.data.get('field_a'))

    @field_a.setter
    def field_a(self, value):
        # 不直接修改data字典，仅记录待更新项
        self._pending_json_updates['$.field_a'] = value
        # 标记实例为脏数据，确保Session会处理它
        session = Session.object_session(self)
        if session:
            session.dirty.add(self)

    @field_a.expression
    def field_a(cls):
        return func.json_extract(cls.data, '$.field_a')
    
    @field_a.update_expression
    def field_a(cls, value):
        return [(cls.data, func.json_set(cls.data, '$.field_a', value))]

    # 同理可以添加field_b的hybrid_property和setter

2. 添加Session事件监听器，处理待更新字段

通过before_flush事件，在Session提交前自动执行json_set的原子更新：

@event.listens_for(Session, 'before_flush')
def handle_pending_json_updates(session, flush_context, instances):
    for instance in list(session.dirty):
        if isinstance(instance, User) and instance._pending_json_updates:
            # 构建批量更新语句，支持同时更新多个JSON字段
            update_stmt = update(User).where(User.id == instance.id)
            current_data_expr = User.data
            for path, value in instance._pending_json_updates.items():
                current_data_expr = func.json_set(current_data_expr, path, value)
            
            # 执行数据库层面的原子更新
            session.execute(update_stmt.values(data=current_data_expr))
            
            # 清除待更新记录，并移除脏标记，避免默认的全列更新
            instance._pending_json_updates.clear()
            session.dirty.discard(instance)

效果验证 #

执行你期望的代码：

user = session.query(User).filter(User.id == 1).one()
user.field_a = 5555
session.commit()

此时数据库执行的SQL是：

UPDATE user_account SET data = json_set(data, '$.field_a', 5555) WHERE id = 1;

该语句会仅修改JSON列中的field_a字段，完全不会影响field_b或其他嵌套字段，即使在查询和更新期间有其他进程修改field_b，也不会造成冲突。

关键优势 #

• 原子性：依赖数据库的json_set函数保证更新操作的原子性，避免并发覆盖问题
• 语法友好：保留了user.field_a = xxx的直观赋值方式，无需编写复杂的update语句
• 批量支持：如果同时更新多个JSON字段（如field_a和field_b），会自动生成链式的json_set调用，一次完成更新

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