SQLAlchemy «Многие ко многим» с использованием on_conflict_do_update PostgreSQL не фиксируется ⇐ Python

[Anonymous]

Я пытаюсь использовать связь «многие ко многим» в SQLAlchemy с PostgreSQL. Я хочу воспользоваться преимуществом on_conflict_do_update, поэтому мне приходится использовать операторы вставки вместо создания объекта ORM, добавления его в сеанс и фиксации.
Я обнаружил, что когда объект создается таким образом, добавление связи «многие ко многим» с помощьюthing.relationship.append не работает. Т.е. при фиксации в таблице связей нет новой строки.
Я воспроизвел проблему ниже с таблицами B и C (и BC для их объединения). Я отметил, как заставить эту связь работать, но вы не можете запустить код дважды, потому что это приведет к ошибке UniqueConstraint.

Код: Выделить всё

from typing import List
from sqlalchemy import Column, create_engine, ForeignKey, String, Table
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session
from sqlalchemy.dialects.postgresql import insert

class Base(DeclarativeBase):
pass
bc_table = Table(
"bc",
Base.metadata,
Column("b_id", ForeignKey("b.id", ondelete="CASCADE"), index=True, primary_key=True),
Column("c_id", ForeignKey("c.id", ondelete="CASCADE"), index=True, primary_key=True)
)
class B(Base):
__tablename__ = 'b'
id: Mapped[int] = mapped_column(primary_key=True)
foo: Mapped[str] = mapped_column(String, unique=True)
cs: Mapped[List["C"]] = relationship(back_populates="bs", secondary=bc_table)
class C(Base):
__tablename__ = 'c'
id: Mapped[int] = mapped_column(primary_key=True)
bar: Mapped[str] = mapped_column(String, unique=True)
bs: Mapped[List["B"]] = relationship(back_populates="cs", secondary=bc_table)

engine = create_engine('postgresql://postgres:hunter2@localhost:5432/replica')
Base.metadata.create_all(engine)
with Session(engine) as session, session.begin():
# I have to use an insert to take advantage of `on_conflict_do_update`
set_ = {'foo': 1}
b_stmt = insert(B).values(set_)
b_stmt = b_stmt.on_conflict_do_update(index_elements=[B.foo], set_=set_)
cro = session.execute(b_stmt)
b = B(id=cro.inserted_primary_key[0], **set_)

set_ = {'bar': 1}
c_stmt = insert(C).values(set_)
c_stmt = c_stmt.on_conflict_do_update(index_elements=[C.bar], set_=set_)
cro = session.execute(c_stmt)
c = C(id=cro.inserted_primary_key[0], **set_)

# Try to create a many-to-many relationship entry between b and c.
c.bs.append(b)

## The easy way would be to do this, but I can't because I need to use on_conflict_do_update
# b = B(foo=1)
# c = C(bar=1)
# c.bs.append(b)
# session.add_all([b, c])

session.commit()

Как заставить c.bs.append зафиксироваться? Является ли решение просто проверить, существуют ли b или c, и создать их только через b = ...; с = ...; session.add_all([b,c])? Я бы предпочел атомарность on_conflict_do_update.


Подробнее здесь: https://stackoverflow.com/questions/790 ... nt-get-com