Как выполнить UPDATE ... JOIN в SQLAlchemy 2.5 без декартового произведения

Показываем, как в SQLAlchemy 2.5 сделать UPDATE ... JOIN в стиле MySQL: внутренний и левый join, фильтр IS NULL и обновление без декартового произведения.

Обновлять строки на основе данных из JOIN — задача встречается часто, но перенести её на SQLAlchemy 2.5 аккуратно, не спровоцировав декартово произведение, бывает непросто. Здесь нам нужен вариант в стиле MySQL: UPDATE ... JOIN с внутренним соединением с таблицей-родителем и левым соединением с третьей таблицей, чтобы обновить отметку времени, когда в третьей таблице нет связанной строки.

Постановка задачи и неудачная попытка

Нужно обновить строки в child, подтянув значение из parent, но только если в other_child нет соответствующей записи. Простой подход с подзапросом приводит к декартовому произведению или ошибкам.

from sqlalchemy import DateTime, ForeignKey, Integer, create_engine, func, select, update
from sqlalchemy.orm import DeclarativeBase, mapped_column, sessionmaker

class ModelBase(DeclarativeBase):
    pass

class ParentRow(ModelBase):
    __tablename__ = "parent"
    id = mapped_column(Integer, primary_key=True, nullable=False)
    updated_at = mapped_column(DateTime, nullable=True)

class ChildRow(ModelBase):
    __tablename__ = "child"
    id = mapped_column(Integer, primary_key=True, nullable=False)
    parent_id = mapped_column(Integer, ForeignKey("parent.id"))
    last_status_change = mapped_column(DateTime, nullable=True)

class SiblingRow(ModelBase):
    __tablename__ = "other_child"
    id = mapped_column(Integer, primary_key=True, nullable=False)

db = create_engine("mysql://root:@127.0.0.1/dev?charset=utf8mb4")
ModelBase.metadata.create_all(db)
SessionFactory = sessionmaker(bind=db)

flt = (
    select(ParentRow.id, ParentRow.updated_at)
    .outerjoin(SiblingRow)
    .where(SiblingRow.id.is_(None))
).subquery()

upd_stmt = (
    update(ChildRow)
    .where(ChildRow.parent_id.in_(select(flt.c.id)))
    .values(last_status_change=func.CONVERT_TZ(flt.c.updated_at, "Europe/Paris", "UTC"))
)

with SessionFactory() as db_sess:
    db_sess.execute(upd_stmt)

Что происходит

Вышеописанный вариант пытается связать UPDATE по child с подзапросом, который отбирает родителей без строк в other_child, а затем использовать этот подзапрос как источник значения для обновления. На деле это не складывается в нужный UPDATE ... JOIN и может привести к декартовому произведению. Целевой SQL должен выглядеть как MySQL-обновление с внутренним и левым JOIN плюс фильтр IS NULL по левой таблице соединения.

Рабочее решение: обновлять напрямую по join-конструкту

Подзапрос не нужен. В SQLAlchemy можно передать в UPDATE непосредственно join-конструкт, и он сгенерирует ровно тот синтаксис MySQL, который требуется.

from sqlalchemy import DateTime, ForeignKey, Integer, create_engine, func, join, update
from sqlalchemy.orm import DeclarativeBase, Session, mapped_column

class ModelBase(DeclarativeBase):
    pass

class ParentRow(ModelBase):
    __tablename__ = "parent"
    id = mapped_column(Integer, primary_key=True, nullable=False)
    updated_at = mapped_column(DateTime, nullable=True)

class ChildRow(ModelBase):
    __tablename__ = "child"
    id = mapped_column(Integer, primary_key=True, nullable=False)
    parent_id = mapped_column(Integer, ForeignKey("parent.id"))
    last_status_change = mapped_column(DateTime, nullable=True)

class SiblingRow(ModelBase):
    __tablename__ = "other_child"
    id = mapped_column(Integer, primary_key=True, nullable=False)
    parent_id = mapped_column(Integer, ForeignKey("parent.id"))

engine = create_engine("mysql+pymysql://", echo=True)
ModelBase.metadata.create_all(engine)

with Session(engine) as db_sess:
    jn = join(
        join(ChildRow, ParentRow, ParentRow.id == ChildRow.parent_id),
        SiblingRow,
        SiblingRow.parent_id == ParentRow.id,
        isouter=True,
    )

    stmt = (
        update(jn)
        .values({
            ChildRow.last_status_change: func.CONVERT_TZ(
                ParentRow.updated_at, "Europe/Paris", "UTC"
            )
        })
        .where(SiblingRow.id.is_(None))
    )

    db_sess.execute(stmt)

В результате получится нужный запрос MySQL.

UPDATE child
INNER JOIN parent ON parent.id = child.parent_id
LEFT OUTER JOIN other_child ON other_child.parent_id = parent.id
SET
    child.last_status_change = CONVERT_TZ(parent.updated_at, %(CONVERT_TZ_1)s, %(CONVERT_TZ_2)s)
WHERE
    other_child.id IS NULL

Почему это важно

Когда бизнес-правило завязано на наличие или отсутствие связанных строк, указание join в качестве цели UPDATE делает выражение ORM эквивалентным желаемому SQL. Это устраняет нежелательные декартовы произведения и повторяет нативную форму MySQL UPDATE ... JOIN, при этом оставаясь в пределах SQLAlchemy без перехода на «сырой» SQL. Кроме того, можно посмотреть сгенерированный SQL через echo у engine или компиляцию выражения — так проще проверить сложные обновления.

Выводы

Если нужно обновлять строки на основе нескольких таблиц, включая левый JOIN и фильтр IS NULL, соберите явный join-конструкт и передайте его в update(). Фильтрацию оставьте на левой таблице соединения, а значения берите напрямую из присоединённых таблиц. Если сомневаетесь, проверьте сгенерированный SQL и убедитесь, что он соответствует задуманному MySQL UPDATE с JOIN.