Verwenden von SQLModel zum Einfügen eines m:n-Beziehungsobjekts, wenn eine Seite der Beziehung bereits in der Datenbank vorhanden ist

Frageninhalt

Ich versuche, SQLModel zu verwenden, um Datensätze in die Datenbank einzufügen, deren Daten wie unten gezeigt sind. Ein Hausobjekt mit einer Farbe und vielen Positionen. Standorte werden auch mit vielen Häusern verknüpft. Geben Sie Folgendes ein:

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

Hier ist ein reproduzierbares Beispiel dafür, was ich versuche:

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))

Das Problem besteht darin, dass beim Ausführen dieses Codes versucht wird, ein doppeltes Standortobjekt zusammen mit dem Hausobjekt einzufügen. Ich hoffe, dass es hier wirklich einfach zu bedienen ist relationship，因为它使访问 house.locations.

Ich kann jedoch nicht herausfinden, wie ich verhindern kann, dass versucht wird, doppelte Positionen einzufügen. Idealerweise hätte ich eine Mapper-Funktion, die die get_or_create-Position übernimmt.

Das Beste, was ich je gesehen habe, sind die zugehörigen Proxys von sqlalchemy. Aber es sieht so aus, als ob SQLModel dies nicht unterstützt.

Weiß jemand, wie man das erreicht? Wenn Sie wissen, wie Sie dies mit sqlalchemy anstelle von sqlmodel erreichen können, wäre ich an Ihrer Lösung interessiert. Ich habe dieses Projekt noch nicht begonnen, daher könnte ich genauso gut sqlalchemy verwenden, wenn es mir das Leben erleichtert.

Ich habe auch versucht, mich anzupassen, indem ich sa_relationship_kwargs wie

verwendet habe

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

Aber dadurch wird verhindert, dass die zugehörigen Einträge zur houselocationlink-Tabelle hinzugefügt werden.

Jeder Hinweis wäre sehr dankbar. Auch wenn das bedeutet, dass ich meine Herangehensweise komplett ändern muss.

Danke!

Richtige Antwort

Ich schreibe diese Lösung, weil Sie erwähnt haben, dass Sie bereit sind, sie zu verwenden sqlalchemy. Wie Sie bereits erwähnt haben, benötigen Sie einen zugehörigen Proxy, aber auch ein „einzigartiges Objekt“. Ich habe dies so angepasst, dass eine asynchrone Abfrage (statt einer Synchronisierung) funktioniert, was meinen persönlichen Vorlieben entspricht, ohne die Logik wesentlich zu ändern.

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())

Sie können feststellen, dass die Behauptung erfolgreich ist, keine eindeutigen Einschränkungen verletzt werden und keine mehrfachen Einfügungen vorliegen. Ich habe einige Inline-Kommentare hinterlassen, in denen die „kritischen“ Aspekte dieses Codes erwähnt werden. Wenn Sie diesen Code mehrmals ausführen, werden Sie feststellen, dass nur das neue house-Objekt und das entsprechende

hinzugefügt werden, nicht jedoch das neue location-Objekt. Für jedes Schlüssel-Wert-Paar wird nur eine Abfrage durchgeführt, um dieses Verhalten zwischenzuspeichern. house 对象和相应的 houselocationlink，而没有添加新的 location

Das obige ist der detaillierte Inhalt vonVerwenden von SQLModel zum Einfügen eines m:n-Beziehungsobjekts, wenn eine Seite der Beziehung bereits in der Datenbank vorhanden ist. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!