SQLAlchemy 多对多关系：UNIQUE 约束失败

Question

所以，我定义了多对多的 SQLAlchemy 关系，

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


mapper_registry = registry()
Base = declarative_base()


bridge_category = Table(
    "bridge_category",
    Base.metadata,
    Column("video_id", ForeignKey("video.id"), primary_key=True),
    Column("category_id", ForeignKey("category.id"), primary_key=True),
    UniqueConstraint("video_id", "category_id"),
)
class BridgeCategory: pass
mapper_registry.map_imperatively(BridgeCategory, bridge_category)


class Video(Base):
    __tablename__ = 'video'

    id = Column(Integer, primary_key=True)
    title = Column(String)
    categories = relationship("Category", secondary=bridge_category, back_populates="videos")


class Category(Base):
    __tablename__ = 'category'

    id = Column(Integer, primary_key=True)
    text = Column(String, unique=True)
    videos = relationship("Video", secondary=bridge_category, back_populates="categories")


engine = create_engine('sqlite:///:memory:', echo=True)
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)

with Session() as s:

    v1 = Video(title='A', categories=[Category(text='blue'), Category(text='red')])
    v2 = Video(title='B', categories=[Category(text='green'), Category(text='red')])
    v3 = Video(title='C', categories=[Category(text='grey'), Category(text='red')])
    videos = [v1, v2, v3]

    s.add_all(videos)
    s.commit()

当然，由于

Category.text

的唯一约束，我们会得到以下错误。

sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) UNIQUE constraint failed: category.text
[SQL: INSERT INTO category (text) VALUES (?) RETURNING id]
[parameters: ('red',)]

我想知道处理这个问题的最佳方法是什么。通过我的程序，我获得了很多视频对象，每个视频对象都有一个唯一的类别对象列表。所有这些视频对象都会发生文本冲突。

我可以循环浏览所有视频和所有类别，形成一个类别集，但这有点蹩脚。我还必须对我的 Video 对象具有的 12 个以上其他多对多关系执行此操作，这似乎效率很低。

我可以为此设置一个“插入忽略”标志吗？我在网上找不到任何有关这种情况的信息。

Answer 1

我提出了一个解决方案，需要在桥模型上进行命名/样式约定，但除此之外，一切都是动态运行的。您可以看到我在下面的 f 字符串中所做的命名假设。

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Table, create_engine, select
from sqlalchemy.orm import relationship, registry, sessionmaker
from sqlalchemy.dialects.sqlite import insert

mapper_registry = registry()
Base = declarative_base()

bridge_category = Table(
    "bridge_category",
    Base.metadata,
    Column("video_id", ForeignKey("video.id"), primary_key=True),
    Column("category_id", ForeignKey("category.id"), primary_key=True),
    UniqueConstraint("video_id", "category_id"),
)
class BridgeCategory: pass
mapper_registry.map_imperatively(BridgeCategory, bridge_category)


class Video(Base):
    __tablename__ = 'video'

    id = Column(Integer, primary_key=True)
    title = Column(String)
    categories = relationship("Category", secondary=bridge_category, back_populates="videos")


class Category(Base):
    __tablename__ = 'category'

    id = Column(Integer, primary_key=True)
    text = Column(String, unique=True)
    videos = relationship("Video", secondary=bridge_category, back_populates="categories")


def get_dict_from_model_obj(obj):
    d = {}
    for column in obj.__table__.columns:
        d[column.name] = getattr(obj, column.name)
    return d


def add_model_object_with_lists(s, obj):
    d = get_dict_from_model_obj(obj)

    d_not_list = {k: v for k, v in d.items() if not isinstance(v, list)} # remove list attrs (.i.e categories)
    model_2 = type(obj)(**d_not_list)
    s.add(model_2)
    s.commit()

    list_models = [obj.__getattribute__(attr) for attr in obj.__dict__ if isinstance(obj.__getattribute__(attr), list)] # get list attrs (i.e. categories)
    for list_model in list_models:
        for model_obj in list_model:

            d = get_dict_from_model_obj(model_obj)
            model_obj_type = type(model_obj)

            sql = insert(model_obj_type).values(d).on_conflict_do_nothing([model_obj_type.text])
            s.execute(sql)
            s.commit()

            model_obj_id = s.scalar(select(model_obj_type.id).where(model_obj_type.text == model_obj.text))

            # makes assumptions about bridge table names and column names
            bridge_class = globals()[f'Bridge{model_obj_type.__name__}']
            sql = insert(bridge_class).values(
                {
                    f'{model_2.__tablename__}_id': model_2.id,
                    f'{model_obj.__tablename__}_id': model_obj_id
                }
            )
            s.execute(sql)
    s.commit()

if __name__=='__main__':
    
    engine = create_engine('sqlite:///:memory:', echo=True)
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)

    v1 = Video(title='A', categories=[Category(text='blue'), Category(text='red')])
    v2 = Video(title='B', categories=[Category(text='green'), Category(text='red')])
    v3 = Video(title='C', categories=[Category(text='grey'), Category(text='red')])
    videos = [v1, v2, v3]

    with Session() as s:
        for video in videos:
            add_model_object_with_lists(s, video)