SQLAlchemy 2.0中ORM方式实现父子表存在则跳过新增逻辑

针对需求——基于name唯一约束实现父、子项的幂等新增（存在则跳过，不存在则创建），以下是纯ORM的实现方案：

1. 模型定义（带唯一约束） #

先确保模型正确添加唯一约束：Parent的name字段唯一，Child的name+parent_id联合唯一（避免同一父项下重复子项）：

from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine

Base = declarative_base()

class Parent(Base):
    __tablename__ = 'parents'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, nullable=False)
    children = relationship("Child", back_populates="parent")
    
    __table_args__ = (
        UniqueConstraint('name', name='uq_parent_name'),
    )

class Child(Base):
    __tablename__ = 'children'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, nullable=False)
    parent_id = Column(Integer, ForeignKey('parents.id'))
    parent = relationship("Parent", back_populates="children")
    
    __table_args__ = (
        UniqueConstraint('name', 'parent_id', name='uq_child_name_parent'),
    )

# 初始化数据库连接
engine = create_engine('postgresql://user:password@localhost/your_db')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

2. 核心实现逻辑 #

核心思路是批量查询已存在的项，避免重复插入，比逐个try-except或merge更高效且符合ORM使用习惯：

步骤1：处理父项，建立名称到实例的映射 #

先批量获取数据库中已有的父项名称，再根据需要创建新父项，同时把新、旧父项存入映射表，方便后续关联子项：

# 模拟要新增的父-子数据结构
new_parent_child_pairs = [
    {"parent_name": "ParentA", "children": ["ChildA1", "ChildA2"]},
    {"parent_name": "ParentB", "children": ["ChildB1", "ChildB2"]},
    {"parent_name": "ParentA", "children": ["ChildA3"]}  # ParentA已存在，仅新增ChildA3
]

# 批量查询已存在的父项名称，减少DB交互
existing_parent_names = {p.name for p in session.query(Parent.name).all()}
parent_map = {}

for pair in new_parent_child_pairs:
    parent_name = pair["parent_name"]
    if parent_name not in existing_parent_names:
        # 创建新父项并加入会话
        new_parent = Parent(name=parent_name)
        session.add(new_parent)
        parent_map[parent_name] = new_parent
    else:
        # 从数据库获取已存在的父项
        existing_parent = session.query(Parent).filter_by(name=parent_name).first()
        parent_map[parent_name] = existing_parent

# 提交父项（可选，提前提交可确保父项ID已生成，方便后续子项查询）
session.commit()

步骤2：处理子项，避免同一父项下重复 #

针对每个父项，先查询其下已存在的子项名称，再新增不存在的子项：

for pair in new_parent_child_pairs:
    current_parent = parent_map[pair["parent_name"]]
    # 批量查询当前父项下已存在的子项名称
    existing_child_names = {c.name for c in session.query(Child.name)
                          .filter_by(parent_id=current_parent.id).all()}
    # 新增不存在的子项
    for child_name in pair["children"]:
        if child_name not in existing_child_names:
            new_child = Child(name=child_name, parent=current_parent)
            session.add(new_child)

# 提交所有子项变更
session.commit()

3. 关键说明 #

• 为什么不用merge()？：merge()是基于主键匹配实例的，而你的主键是自增ID，创建实例时无法提前获取，因此merge()会默认将所有实例视为新对象尝试插入，触发唯一约束报错。
• 性能优化：批量查询已存在的名称（而非逐个查询）能大幅减少数据库交互次数，适合数据量较大的场景。
• 替代方案（混合Core+ORM）：如果接受少量Core语法，也可以用Session.execute配合on conflict do nothing实现，但这不属于纯ORM方式，示例如下（供参考）：

  # 父项批量插入（Core+ORM混合）
  session.execute(
      Parent.__table__.insert().values([{"name": "ParentA"}, {"name": "ParentB"}])
      .on_conflict_do_nothing(constraint='uq_parent_name')
  )
  session.commit()
  

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