如何基于JSON Patch更新PostgreSQL中的嵌套关联数据

架构老林

2026-7-7

问题描述

每隔几小时会获取一份包含Game对象的复杂JSON数据，Game下有Player列表，每个Player又包含Training列表（各对象还带有整数、字符串等其他字段）。需求如下：

若Game对象不存在于PostgreSQL数据库（通过unique_id校验），则将整个结构插入Game、Player、Training对应表；
若Game已存在，则需要更新对应数据。

当前可获取原始JSON、更新后JSON及JSON Patch，但遇到两个核心问题：

数据库中查询出的列表（如Player列表）顺序与updated_object中的列表顺序不一致；
需要依托数据库主键让ORM识别待更新对象，无法直接通过JSON Patch应用到数据库转成的JSON上。

模型定义

class Game(Base):
    __tablename__ = "game"
    game_id: int = Column(INTEGER, primary_key=True,
                              server_default=Identity(always=True, start=1, increment=1, minvalue=1,
                                                      maxvalue=2147483647, cycle=False, cache=1),
                              autoincrement=True)
    unique_id: str = Column(TEXT, nullable=False)
    name: str = Column(TEXT, nullable=False)
    players = relationship('Player', back_populates='game')

class Player(Base):
    __tablename__ = "player"
    player_id: int = Column(INTEGER, primary_key=True,
                          server_default=Identity(always=True, start=1, increment=1, minvalue=1,
                                                  maxvalue=2147483647, cycle=False, cache=1),
                          autoincrement=True)
    unique_id: str = Column(TEXT, nullable=False)
    game_id: int = Column(INTEGER, ForeignKey('game.game_id'), nullable=False)
    name: str = Column(TEXT, nullable=False)
    birth_date = Column(DateTime, nullable=False)
    game = relationship('Game', back_populates='players')
    trainings = relationship('Training', back_populates='player')

class Training(Base):
    __tablename__ = "training"
    training_id: int = Column(INTEGER, primary_key=True,
                               server_default=Identity(always=True, start=1, increment=1, minvalue=1,
                                                       maxvalue=2147483647, cycle=False, cache=1),
                               autoincrement=True)
    unique_id: str = Column(TEXT, nullable=False)
    name: str = Column(TEXT, nullable=False)
    number_of_players: int = Column(INTEGER, nullable=False)
    player_id: int = Column(INTEGER, ForeignKey('player.player_id'), nullable=False)
    player = relationship('Player', back_populates='players')

更新数据示例JSON

{"original_object":{"name":"Table Tennis","unique_id":"432","players":[{"unique_id":"793","name":"John","birth_date":"2023-10-28T00:10:56Z","trainings":[{"unique_id":"43","name":"Morning Session","number_of_players":3}, {"unique_id":"44","name":"Evening Session","number_of_players":2}]}]},"updated_object":{"name":"Table Tennis","unique_id":"432","players":[{"unique_id":"793","name":"John","birth_date":"2023-10-28T00:10:56Z","trainings":[{"unique_id":"43","name":"Morning Session","number_of_players":3}, {"unique_id":"44","name":"Evening Session","number_of_players":4}]}]},"json_patch":[{"op":"replace","path":"/players/0/trainings/1/numbre_of_players","value":4}],"timestamp":"2023-10-28T02:00:36Z"}

注：该JSON Patch意图将第二个Training的number_of_players字段更新为4（Patch中字段名存在笔误numbre_of_players）

现有新增Game代码

Session = sessionmaker(bind=engine_sync)
session = Session()
session.begin()
game = Game.from_dict(json['updated_object'])
existing_game = session.query(Game).filter_by(unique_id=game.id).first()
if not existing_game:
    session.add(game)
    session.commit()

最优处理方案

核心思路：基于`unique_id`匹配对象，而非列表索引，ORM层面直接更新关联对象

不需要将数据库数据转成JSON再应用Patch，而是直接通过unique_id定位到数据库中的Game、Player、Training实例，再用updated_object中的数据覆盖更新，同时处理新增/删除的子对象。

1. 封装通用的对象更新方法

为每个模型类添加一个update_from_dict方法，用于将字典中的字段值更新到实例上（跳过主键、外键等不需要更新的字段）：

class Game(Base):
    # 原有模型定义...
    def update_from_dict(self, data):
        # 更新非关联字段
        for key, value in data.items():
            if key not in ['game_id', 'players']:
                setattr(self, key, value)

class Player(Base):
    # 原有模型定义...
    def update_from_dict(self, data):
        for key, value in data.items():
            if key not in ['player_id', 'game_id', 'trainings']:
                setattr(self, key, value)

class Training(Base):
    # 原有模型定义...
    def update_from_dict(self, data):
        for key, value in data.items():
            if key not in ['training_id', 'player_id']:
                setattr(self, key, value)

2. 完整的Game新增/更新逻辑

当Game已存在时，按以下步骤处理：

用updated_object中的基础字段更新Game实例；
遍历updated_object中的Player列表，通过unique_id匹配数据库中已有的Player：
- 匹配到则调用update_from_dict更新Player字段，再处理其Training列表；
- 未匹配到则创建新Player实例，关联到当前Game并添加到会话；
清理数据库中存在但updated_object里没有的Player（如果需要支持删除）；
同理处理每个Player下的Training列表。

完整代码示例：

Session = sessionmaker(bind=engine_sync)
session = Session()
try:
    session.begin()
    updated_game_data = json['updated_object']
    # 查找现有Game
    existing_game = session.query(Game).filter_by(unique_id=updated_game_data['unique_id']).first()
    
    if not existing_game:
        # 新增逻辑
        game = Game.from_dict(updated_game_data)
        session.add(game)
    else:
        # 更新Game基础字段
        existing_game.update_from_dict(updated_game_data)
        
        # 处理Player列表：构建现有Player的unique_id映射
        existing_players = {p.unique_id: p for p in existing_game.players}
        updated_player_dicts = updated_game_data['players']
        
        for player_dict in updated_player_dicts:
            player_unique_id = player_dict['unique_id']
            if player_unique_id in existing_players:
                # 更新现有Player
                player = existing_players.pop(player_unique_id)
                player.update_from_dict(player_dict)
                
                # 处理该Player的Training列表：构建现有Training的unique_id映射
                existing_trainings = {t.unique_id: t for t in player.trainings}
                updated_training_dicts = player_dict['trainings']
                
                for training_dict in updated_training_dicts:
                    training_unique_id = training_dict['unique_id']
                    if training_unique_id in existing_trainings:
                        # 更新现有Training
                        training = existing_trainings.pop(training_unique_id)
                        training.update_from_dict(training_dict)
                    else:
                        # 新增Training
                        new_training = Training.from_dict(training_dict)
                        new_training.player = player
                        session.add(new_training)
                
                # 删除数据库中存在但更新数据里没有的Training
                for training in existing_trainings.values():
                    session.delete(training)
            else:
                # 新增Player及关联的Training
                new_player = Player.from_dict(player_dict)
                new_player.game = existing_game
                session.add(new_player)
                for training_dict in player_dict['trainings']:
                    new_training = Training.from_dict(training_dict)
                    new_training.player = new_player
                    session.add(new_training)
        
        # 删除数据库中存在但更新数据里没有的Player
        for player in existing_players.values():
            session.delete(player)
    
    session.commit()
except Exception as e:
    session.rollback()
    raise e
finally:
    session.close()

3. 可选：利用JSON Patch优化更新（需Patch可靠）

如果JSON Patch的字段路径可以转换为基于unique_id的定位（而非依赖列表索引），可以直接解析Patch更新实例：

修复Patch中的字段名错误（如示例中的numbre_of_players）；
将路径中的索引（如/players/0/trainings/1）转换为通过unique_id查找对象：先找到Game，再匹配对应unique_id的Player，再匹配对应unique_id的Training，最后更新字段。
但这种方式需要额外的路径解析逻辑，不如直接用updated_object全量更新（只要updated_object是完整的最新数据）更可靠。

关键优势

完全基于unique_id匹配对象，不受列表顺序影响；
ORM直接操作实例，自动处理主键和外键关联；
统一处理新增、更新、删除三种场景；
逻辑清晰，易于维护和扩展。

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

如何基于JSON Patch更新PostgreSQL中的嵌套关联数据

模型定义 #

更新数据示例JSON #

现有新增Game代码 #

最优处理方案 #

核心思路：基于unique_id匹配对象，而非列表索引，ORM层面直接更新关联对象 #