先创建一个基类,用来表示某个类支持逻辑删除
class SoftDeleteModel: '''逻辑删除基类 用来实现逻辑删除。 继承这个基类的子类需要在数据库的列中存在deleted_at列,类型为varchar。 ''' deleted_at: Mapped[str] = mapped_column(String(50), default=None)
模型类如果要支持逻辑删除需继承SoftDeleteModel:
class User(Base,SoftDeleteModel): __tablename__ = "user" id:Mapped[int]=mapped_column(autoincrement=True,primary_key=True) phone_number:Mapped[str]=mapped_column(String(11)) ...
添加sqlalchemy 事件监听器,拦截每次select操作,自动执行过滤软删除操作
@event.listens_for(Session, "do_orm_execute") def _add_filtering_deleted_at(execute_state): """ 自动过滤掉被软删除的数据。 deleted_at不为null即为被软删除。 使用以下方法可以获得被软删除的数据。 query(...).filter(...).execution_options(include_deleted=True) """ if ( execute_state.is_select and not execute_state.is_column_load and not execute_state.is_relationship_load and not execute_state.execution_options.get("include_deleted", False) ): execute_state.statement = execute_state.statement.options( with_loader_criteria( SoftDeleteModel, lambda cls: cls.deleted_at.is_(None), # deleted_at列为空则为未被软删除 include_aliases=True, ) )
标签:execute,sqlalchemy,删除,deleted,SoftDeleteModel,state,过滤,True From: https://www.cnblogs.com/axel10/p/17608474.html