當關係的一側已存在於資料庫中時，使用 SQLModel 插入多對多關係對象

問題內容

我正在嘗試使用 sqlmodel 在資料庫中插入記錄，其中資料如下所示。 一個 house 對象，它有顏色和許多位置。 地點也將與許多房屋相關聯。輸入為：

[
    {
        "color": "red",
        "locations": [
            {"type": "country", "name": "netherlands"},
            {"type": "municipality", "name": "amsterdam"},
        ],
    },
    {
        "color": "green",
        "locations": [
            {"type": "country", "name": "netherlands"},
            {"type": "municipality", "name": "amsterdam"},
        ],
    },
]

這是我正在嘗試做的事情的可重現範例：

import asyncio
from typing import list

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlmodel import field, relationship, sqlmodel, uniqueconstraint
from sqlmodel.ext.asyncio.session import asyncsession

database_url = "sqlite+aiosqlite:///./database.db"


engine = create_async_engine(database_url, echo=true, future=true)


async def init_db() -> none:
    async with engine.begin() as conn:
        await conn.run_sync(sqlmodel.metadata.create_all)


sessionlocal = sessionmaker(
    autocommit=false,
    autoflush=false,
    bind=engine,
    class_=asyncsession,
    expire_on_commit=false,
)


class houselocationlink(sqlmodel, table=true):
    house_id: int = field(foreign_key="house.id", nullable=false, primary_key=true)
    location_id: int = field(
        foreign_key="location.id", nullable=false, primary_key=true
    )


class location(sqlmodel, table=true):
    id: int = field(primary_key=true)
    type: str  # country, county, municipality, district, city, area, street, etc
    name: str  # amsterdam, germany, my street, etc

    houses: list["house"] = relationship(
        back_populates="locations",
        link_model=houselocationlink,
    )

    __table_args__ = (uniqueconstraint("type", "name"),)


class house(sqlmodel, table=true):
    id: int = field(primary_key=true)
    color: str = field()
    locations: list["location"] = relationship(
        back_populates="houses",
        link_model=houselocationlink,
    )
    # other fields...


data = [
    {
        "color": "red",
        "locations": [
            {"type": "country", "name": "netherlands"},
            {"type": "municipality", "name": "amsterdam"},
        ],
    },
    {
        "color": "green",
        "locations": [
            {"type": "country", "name": "netherlands"},
            {"type": "municipality", "name": "amsterdam"},
        ],
    },
]


async def add_houses(payload) -> list[house]:
    result = []
    async with sessionlocal() as session:
        for item in payload:
            locations = []
            for location in item["locations"]:
                locations.append(location(**location))
            house = house(color=item["color"], locations=locations)
            result.append(house)
        session.add_all(result)
        await session.commit()


asyncio.run(init_db())
asyncio.run(add_houses(data))

問題是，當我運行此程式碼時，它嘗試將重複的位置物件與房屋物件一起插入。 我希望能夠在這裡使用 relationship，因為它使存取 house.locations 變得非常容易。

但是，我無法弄清楚如何阻止它嘗試插入重複的位置。理想情況下，我有一個映射器函數來執行 get_or_create 位置。

我所見過的最能實現這一點的是 sqlalchemy 的關聯代理。但看起來 sqlmodel 不支援這一點。

有人知道如何實現這一目標嗎？如果您知道如何使用 sqlalchemy 而不是 sqlmodel 來完成此操作，我有興趣查看您的解決方案。我還沒有開始這個項目，所以如果它能讓我的生活更輕鬆的話，我不妨使用 sqlalchemy。

我還嘗試使用 sa_relationship_kwargs 進行調整，例如

sa_relationship_kwargs={
    "lazy": "selectin",
    "cascade": "none",
    "viewonly": "true",
}

但這會阻止將關聯條目新增至 houselocationlink 表中。

任何指示將不勝感激。即使這意味著完全改變我的方法。

謝謝！

正確答案

我正在寫這個解決方案，因為您提到您願意使用 sqlalchemy。正如您所提到的，您需要關聯代理，但您還需要「唯一物件」。我已將其調整為非同步查詢（而不是同步）功能，與我的個人偏好保持一致，所有這些都沒有顯著改變邏輯。

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 物件。對於每個鍵值對，只會進行一次查詢來快取此行為。

以上是當關係的一側已存在於資料庫中時，使用 SQLModel 插入多對多關係對象的詳細內容。更多資訊請關注PHP中文網其他相關文章！