Menggunakan SQLModel untuk memasukkan objek perhubungan banyak-ke-banyak apabila satu bahagian perhubungan sudah wujud dalam pangkalan data

Kandungan soalan

Saya cuba menggunakan sqlmodel untuk memasukkan rekod dalam pangkalan data di mana data adalah seperti yang ditunjukkan di bawah. Objek rumah dengan warna dan banyak lokasi. Lokasi juga akan dikaitkan dengan banyak rumah. Masukkan sebagai:

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

Berikut ialah contoh yang boleh diterbitkan semula tentang perkara yang saya cuba lakukan:

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

Masalahnya ialah apabila saya menjalankan kod ini, ia cuba memasukkan objek lokasi pendua bersama objek rumah. Saya harap ia menjadi sangat mudah untuk digunakan relationship，因为它使访问 house.locations di sini.

Namun, saya tidak dapat memikirkan cara untuk menghalangnya daripada cuba memasukkan kedudukan pendua. Sebaik-baiknya, saya akan mempunyai fungsi pemeta yang melakukan kedudukan get_or_create.

Yang terbaik yang saya lihat yang melakukan ini ialah proksi berkaitan sqlalchemy. Tetapi nampaknya sqlmodel tidak menyokong ini.

Adakah sesiapa tahu bagaimana untuk mencapai ini? Jika anda tahu cara untuk mencapai ini menggunakan sqlalchemy dan bukannya sqlmodel, saya akan berminat untuk melihat penyelesaian anda. Saya belum memulakan projek ini lagi, jadi saya mungkin menggunakan sqlalchemy jika ia menjadikan hidup saya lebih mudah.

Saya pun cuba adjust guna sa_relationship_kwargs like

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

Tetapi ini akan menghalang entri yang berkaitan daripada ditambahkan pada jadual houselocationlink.

Sebarang petunjuk akan sangat dihargai. Walaupun ia bermakna mengubah pendekatan saya sepenuhnya.

Terima kasih!

Jawapan betul

Saya menulis penyelesaian ini kerana anda menyebut bahawa anda sanggup menggunakan sqlalchemy. Seperti yang anda nyatakan, anda memerlukan proksi yang berkaitan, tetapi anda juga memerlukan "objek unik". Saya telah melaraskan ini untuk berfungsi pertanyaan async (bukannya segerak), selaras dengan keutamaan peribadi saya, semuanya tanpa mengubah logik dengan ketara.

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

Anda dapat melihat bahawa penegasan itu lulus, tiada kekangan unik dilanggar, dan tiada berbilang sisipan. Saya telah meninggalkan beberapa komen sebaris yang menyebut aspek "kritikal" kod ini. Jika anda menjalankan kod ini beberapa kali, anda akan dapati bahawa hanya objek house baharu dan house 对象和相应的 houselocationlink，而没有添加新的 location yang sepadan ditambah, tetapi bukan objek location baharu. Hanya satu pertanyaan dibuat bagi setiap pasangan nilai kunci untuk menyimpan kelakuan ini.

Atas ialah kandungan terperinci Menggunakan SQLModel untuk memasukkan objek perhubungan banyak-ke-banyak apabila satu bahagian perhubungan sudah wujud dalam pangkalan data. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!