使用SQLModel插入多对多关系对象时避免重复关联已存在数据

方案一：继续使用SQLModel，通过业务逻辑实现get_or_create #

核心思路是在关联House和Location前，先检查数据库中是否已存在目标Location，存在则直接复用实例，不存在再创建，同时给Location的唯一标识字段加数据库约束确保数据唯一性。

1. 定义完整模型（含关联表） #

from sqlmodel import SQLModel, Field, Relationship, Session, create_engine

# 多对多关联表
class HouseLocationLink(SQLModel, table=True):
    house_id: int | None = Field(default=None, foreign_key="house.id", primary_key=True)
    location_id: int | None = Field(default=None, foreign_key="location.id", primary_key=True)

class Location(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(unique=True)  # 唯一约束，从数据库层面防止重复
    houses: list["House"] = Relationship(back_populates="locations", link_model=HouseLocationLink)

class House(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    color: str
    locations: list[Location] = Relationship(back_populates="houses", link_model=HouseLocationLink)

# 初始化数据库
engine = create_engine("sqlite:///house_db.db")
SQLModel.metadata.create_all(engine)

2. 实现get_or_create_location工具函数 #

def get_or_create_location(session: Session, name: str) -> Location:
    # 先查询是否存在对应名称的Location
    existing_loc = session.query(Location).filter(Location.name == name).first()
    if existing_loc:
        return existing_loc
    # 不存在则创建并保存
    new_loc = Location(name=name)
    session.add(new_loc)
    session.commit()
    session.refresh(new_loc)
    return new_loc

3. 实际使用示例 #

with Session(engine) as session:
    # 获取或创建Location实例
    beijing = get_or_create_location(session, "北京")
    shanghai = get_or_create_location(session, "上海")
    
    # 创建House并关联已存在的Location
    white_house = House(color="白色", locations=[beijing, shanghai])
    session.add(white_house)
    session.commit()
    
    # 再次创建House复用相同Location，不会重复插入
    black_house = House(color="黑色", locations=[get_or_create_location(session, "北京")])
    session.add(black_house)
    session.commit()
    
    # 验证relationship访问正常
    saved_house = session.query(House).first()
    print(saved_house.locations)  # 输出关联的Location列表

方案二：改用SQLAlchemy，结合association proxy简化逻辑 #

SQLAlchemy支持association_proxy，可以直接通过名称管理关联的Location，底层自动复用已有实例，同时保留locations关系用于直接访问对象。

1. 定义模型与关联逻辑 #

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

Base = declarative_base()

# 关联表
class HouseLocationLink(Base):
    __tablename__ = "house_location_link"
    house_id = Column(Integer, ForeignKey("house.id"), primary_key=True)
    location_id = Column(Integer, ForeignKey("location.id"), primary_key=True)
    location = relationship("Location")

    def __init__(self, location):
        self.location = location

class Location(Base):
    __tablename__ = "location"
    id = Column(Integer, primary_key=True)
    name = Column(String, unique=True)
    houses = relationship("House", secondary="house_location_link", back_populates="locations")

class House(Base):
    __tablename__ = "house"
    id = Column(Integer, primary_key=True)
    color = Column(String)
    
    # 关联到中间表
    location_links = relationship("HouseLocationLink", cascade="all, delete-orphan")
    # 使用association_proxy直接通过名称管理Location
    location_names = association_proxy(
        "location_links",
        "location",
        creator=lambda name: HouseLocationLink(get_or_create_location(session, name))
    )
    # 保留原始relationship用于直接访问Location实例
    locations = relationship("Location", secondary="house_location_link", back_populates="houses")

# 初始化数据库
engine = create_engine("sqlite:///house_sqlalchemy.db")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

2. 复用get_or_create_location函数 #

def get_or_create_location(session, name):
    existing_loc = session.query(Location).filter(Location.name == name).first()
    if existing_loc:
        return existing_loc
    new_loc = Location(name=name)
    session.add(new_loc)
    session.commit()
    session.refresh(new_loc)
    return new_loc

3. 使用示例 #

# 直接传入Location名称，自动处理复用/创建
blue_house = House(color="蓝色", location_names=["北京", "广州"])
session.add(blue_house)
session.commit()

# 重复使用相同名称不会插入新Location
red_house = House(color="红色", location_names=["北京"])
session.add(red_house)
session.commit()

# 验证relationship访问正常
saved_house = session.query(House).first()
print(saved_house.locations)

核心思路总结 #

• 避免重复插入的核心是先查询再复用/创建关联对象，关系配置（如sa_relationship_kwargs）不负责处理实例存在性，需在业务逻辑层实现。
• 给关联对象的唯一标识字段（如Location.name）加数据库唯一约束，确保数据一致性，防止意外重复插入。
• SQLModel适合轻量场景，手动处理get_or_create即可；SQLAlchemy的association_proxy可进一步简化关联对象的管理逻辑。

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