Python 当关系的一方已经存在于数据库中时，使用SQLModel插入多对多关系对象

我之所以写这个解决方案，是因为你提到你对使用SQLAlchemy持开放态度.正如您所提到的，您需要关联代理，但您还需要"Unique Objects".我对它进行了调整，使其能够处理异步查询(而不是同步查询)，与我的个人偏好保持一致，所有这些都不会显著改变逻辑.

import asyncio
from sqlalchemy import UniqueConstraint, ForeignKey, select, text, func
from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped, relationship
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy

class Base(DeclarativeBase):
    pass

class UniqueMixin:
    cache = {}

    @classmethod
    async def as_unique(cls, session: AsyncSession, *args, **kwargs):
        key = cls, cls.unique_hash(*args, **kwargs)
        if key in cls.cache:
            return cls.cache[key]
        with session.no_autoflush:
            statement = select(cls).where(cls.unique_filter(*args, **kwargs)).limit(1)
            obj = (await session.scalars(statement)).first()
            if obj is None:
                obj = cls(*args, **kwargs)
                session.add(obj)
        cls.cache[key] = obj
        return obj

    @classmethod
    def unique_hash(cls, *args, **kwargs):
        raise NotImplementedError("Implement this in subclass")

    @classmethod
    def unique_filter(cls, *args, **kwargs):
        raise NotImplementedError("Implement this in subclass")

class Location(UniqueMixin, Base):
    __tablename__ = "location"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    type: Mapped[str] = mapped_column()
    house_associations: Mapped[list["HouseLocationLink"]] = relationship(back_populates="location")
    __table_args = (UniqueConstraint(type, name),)

    @classmethod
    def unique_hash(cls, name, type):
        # this is the key for the dict
        return type, name

    @classmethod
    def unique_filter(cls, name, type):
        # this is how you want to establish the uniqueness
        # the result of this filter will be the value in the dict
        return (cls.type == type) & (cls.name == name)

class House(Base):
    __tablename__ = "house"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    location_associations: Mapped[list["HouseLocationLink"]] = relationship(back_populates="house")
    locations: AssociationProxy[list[Location]] = association_proxy(
        "location_associations",
        "location",
        # you need this so you can directly add ``Location`` objects to ``House``
        creator=lambda location: HouseLocationLink(location=location),
    )

class HouseLocationLink(Base):
    __tablename__ = "houselocationlink"
    house_id: Mapped[int] = mapped_column(ForeignKey(House.id), primary_key=True)
    location_id: Mapped[int] = mapped_column(ForeignKey(Location.id), primary_key=True)
    location: Mapped[Location] = relationship(back_populates="house_associations")
    house: Mapped[House] = relationship(back_populates="location_associations")

engine = create_async_engine("sqlite+aiosqlite:///test.sqlite")

async def main():
    data = [
        {
            "name": "red",
            "locations": [
                {"type": "country", "name": "Netherlands"},
                {"type": "municipality", "name": "Amsterdam"},
            ],
        },
        {
            "name": "green",
            "locations": [
                {"type": "country", "name": "Netherlands"},
                {"type": "municipality", "name": "Amsterdam"},
            ],
        },
    ]

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async with AsyncSession(engine) as session, session.begin():
        for item in data:
            house = House(
                name=item["name"],
                locations=[await Location.as_unique(session, **location) for location in item["locations"]]
            )
            session.add(house)

    async with AsyncSession(engine) as session:
        statement = select(func.count(text("*")), Location)
        assert await session.scalar(statement) == 2

        statement = select(func.count(text("*")), House)
        assert await session.scalar(statement) == 2

        statement = select(func.count(text("*")), HouseLocationLink)
        assert await session.scalar(statement) == 4


asyncio.run(main())

您可以注意到，断言确实通过了，没有违反唯一约束，也没有多次插入.我留下了一些内联注释，其中提到了这段代码的"关键"方面.如果您多次运行此代码，您会注意到只添加了新的House个对象和相应的HouseLocationLink个对象，没有添加新的Location个对象.对于每个键-值对，将只进行一个查询来缓存此行为.