Soft deletion via do_orm_execute

[redb0]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from datetime import datetime
from typing import List, Optional
import sqlalchemy as sa
from pydantic import ConfigDict
from sqlmodel import Field, Relationship, SQLModel, Session, col, select
from sqlalchemy.orm import ORMExecuteState, with_loader_criteria, joinedload, selectinload


class Base(SQLModel):
    """Базовый класс моделей."""

    model_config = ConfigDict(populate_by_name=True)


class DeletedAt(Base):
    deleted_at: Optional[datetime] = Field(
        default=None,
        sa_type=sa.DateTime(timezone=True),
        description='Дата и время удаления записи',
    )

    @classmethod
    def make_where_criteria(cls) -> sa.BinaryExpression[bool]:
        return col(cls.deleted_at).is_(None)


class Hero(DeletedAt, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(title='Hero name')


class Team(DeletedAt, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(title='Team name')

    hero_links: List['HeroTeamLink'] = Relationship(
        back_populates='team',
        sa_relationship_kwargs={
            'lazy': 'joined',
        },
    )


class HeroTeamLink(DeletedAt, table=True):
    hero_id: Optional[int] = Field(
        title='Hero ID',
        default=None,
        foreign_key='hero.id',
        primary_key=True,
    )
    team_id: Optional[int] = Field(
        title='Team ID',
        default=None,
        foreign_key='team.id',
        primary_key=True,
    )
    role: str = Field(
        title='Hero role',
        primary_key=True,
    )

    hero: Hero = Relationship(
        sa_relationship_kwargs={
            'lazy': 'joined',
        },
    )
    team: Team = Relationship(
        back_populates='hero_links',
        sa_relationship_kwargs={
            'lazy': 'joined',
        },
    )


@sa.event.listens_for(Session, 'do_orm_execute')
def _do_orm_execute(orm_execute_state: ORMExecuteState) -> None:
    # https://docs.sqlalchemy.org/en/20/orm/session_events.html
    # https://docs.sqlalchemy.org/en/20/_modules/examples/extending_query/filter_public.html
    if not (
        orm_execute_state.is_select
        and not orm_execute_state.is_column_load
        and not orm_execute_state.is_relationship_load
    ):
        return

    orm_execute_state.statement = orm_execute_state.statement.options(
        with_loader_criteria(
            DeletedAt,
            lambda cls: cls.make_where_criteria() if hasattr(cls, 'deleted_at') else None,
            include_aliases=True,
            propagate_to_loaders=False,
        ),
    )


engine = sa.create_engine("postgresql://postgres:postgres@localhost:5432/postgres", echo=True)


def main():
    with Session(engine) as session:
        Base.metadata.create_all(engine)

    with Session(engine) as session:
        hero_1 = Hero(name='Foo')
        hero_2 = Hero(name='Bar')
        hero_3 = Hero(name='Baz')
        session.add(hero_1)
        session.add(hero_2)
        session.add(hero_3)

        team = Team(name='Team A')
        session.add(team)

        link_1 = HeroTeamLink(hero=hero_1, team=team, role='support')
        link_2 = HeroTeamLink(hero=hero_2, team=team, role='damager')
        link_3 = HeroTeamLink(hero=hero_3, team=team, role='tank')
        session.add(link_1)
        session.add(link_2)
        session.add(link_3)

        session.commit()

    with Session(engine) as session:
        teams = session.exec(select(Team)).unique().all()
        for team_ in teams:
            print(f'{team_ = }')
            for hero_ in team_.hero_links:
                print(f'{hero_ = }')
        print('-' * 50)

    with Session(engine) as session:
        team = session.get(Team, 1)
        print(f'{team}; {team.hero_links}')
        team.hero_links[0].deleted_at = datetime.now()
        session.add(team)
        session.commit()
        session.refresh(team, ['hero_links'])

        print(f'{team = }')
        for hero_ in team.hero_links:  # <-- why this 3 links?
            print(f'{hero_ = }')

        # options=[joinedload(Team.hero_links)] -> returning 3 links
        team_ = session.get(Team, 1, options=[selectinload(Team.hero_links)])
        for hero_ in team_.hero_links:
            print(f'{hero_ = }')

    with Session(engine) as session:
        Base.metadata.drop_all(engine)

main()

Description

I'm trying to implement a soft delete:

• https://docs.sqlalchemy.org/en/20/orm/session_events.html
• https://docs.sqlalchemy.org/en/20/_modules/examples/extending_query/filter_public.html

Everything is fine with ordinary objects. But when using associative tables, it is not possible to filter out connected objects. The filtering condition is added only to the Hero table, but not to the HeroTeamLink. Tell me, what am I doing wrong?

Operating System

Linux

Operating System Details

No response

SQLModel Version

0.0.16

Python Version

3.9.19

Additional Context

No response

[jaeyunha]

have you solved all the problem?
I am trying to implement soft-delete and followed your way, and I keep getting an error "~~ column reference "deleted_at" is ambiguous" when running db works with related tables. is it because I don't have sa_relationship_kwargs={
'lazy': 'joined',
}
in my tables for relationship? would you know why?

[redb0]

Note that in my DeletedAt example, there is no table=true attribute.

[redb0]

I used the propagate_to_loaders argument incorrectly. Its meaning should be true, not false.

See more info