Utilisation de SQLModel pour insérer un objet relation plusieurs-à-plusieurs lorsqu'un côté de la relation existe déjà dans la base de données

Contenu de la question

J'essaie d'utiliser sqlmodel pour insérer des enregistrements dans la base de données où les données sont comme indiqué ci-dessous. Un objet de maison avec une couleur et de nombreuses positions. Les emplacements seront également associés à de nombreuses maisons. Entrez comme :

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

Voici un exemple reproductible de ce que j'essaie de faire :

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

Le problème est que lorsque j'exécute ce code, il essaie d'insérer un objet de localisation en double avec l'objet maison. J'espère que cela deviendra vraiment facile à utiliser relationship，因为它使访问 house.locations ici.

Cependant, je n'arrive pas à comprendre comment l'empêcher d'essayer d'insérer des positions en double. Idéalement, j'aurais une fonction mappeur qui fait la position get_or_create.

Le meilleur que j'ai vu pour faire cela, ce sont les proxys associés à sqlalchemy. Mais il semble que sqlmodel ne prenne pas en charge cela.

Est-ce que quelqu'un sait comment y parvenir ? Si vous savez comment y parvenir en utilisant sqlalchemy au lieu de sqlmodel, je serais intéressé de voir votre solution. Je n'ai pas encore commencé ce projet, donc autant utiliser sqlalchemy si cela me facilite la vie.

J'ai aussi essayé de m'ajuster en utilisant sa_relationship_kwargs comme

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

Mais cela empêchera que les entrées associées soient ajoutées au tableau houselocationlink.

Tous les conseils seraient grandement appréciés. Quitte à changer complètement d’approche.

Merci !

Bonne réponse

J'écris cette solution parce que vous avez mentionné que vous êtes prêt à utiliser sqlalchemy. Comme vous l'avez mentionné, vous avez besoin d'un proxy associé, mais vous avez également besoin d'un "objet unique". Je l'ai adapté pour fonctionner comme des requêtes asynchrones (plutôt que synchrones), conformément à mes préférences personnelles, le tout sans changer de manière significative la logique.

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

Vous pouvez remarquer que l'assertion réussit, qu'aucune contrainte unique n'est violée et qu'il n'y a pas d'insertions multiples. J'ai laissé quelques commentaires en ligne mentionnant les aspects « critiques » de ce code. Si vous exécutez ce code plusieurs fois, vous remarquerez que seul le nouvel objet house et le house 对象和相应的 houselocationlink，而没有添加新的 location correspondant sont ajoutés, mais pas le nouvel objet location. Une seule requête est effectuée pour chaque paire clé-valeur afin de mettre en cache ce comportement.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!