From de3081b87a239bf3f52eec0556b7a2db7b78d89d Mon Sep 17 00:00:00 2001 From: Wu Clan Date: Fri, 29 May 2026 17:04:49 +0800 Subject: [PATCH] Add logical deletion custom column value --- docs/usage/crud.md | 16 +++++++++++++++- sqlalchemy_crud_plus/crud.py | 4 +++- tests/models/basic.py | 2 ++ tests/test_delete.py | 30 ++++++++++++++++++++++++++++++ 4 files changed, 50 insertions(+), 2 deletions(-) diff --git a/docs/usage/crud.md b/docs/usage/crud.md index 565dfff..0bf77c6 100644 --- a/docs/usage/crud.md +++ b/docs/usage/crud.md @@ -202,7 +202,7 @@ count = await user_crud.delete_model_by_column( ) ``` -逻辑删除会把标记字段设为 `True`。如果模型包含删除时间字段，也会写入当前时间。 +逻辑删除默认会把标记字段设为 `True`。如果模型包含删除时间字段，也会写入当前时间。 ```python count = await user_crud.delete_model_by_column( @@ -214,6 +214,20 @@ count = await user_crud.delete_model_by_column( ) ``` +如果逻辑删除标记需要写入自定义值，可以使用 `deleted_flag_value`。该参数也支持 SQLAlchemy 表达式，例如把 `deleted` 字段写成当前行的 `id`。 + +```python +count = await user_crud.delete_model_by_column( + session, + logical_deletion=True, + deleted_flag_column='deleted', + deleted_flag_value=User.id, + deleted_at_column='deleted_time', + id=1, + deleted=0 +) +``` + ## 统计和存在性 ```python diff --git a/sqlalchemy_crud_plus/crud.py b/sqlalchemy_crud_plus/crud.py index 04327f6..17174c3 100644 --- a/sqlalchemy_crud_plus/crud.py +++ b/sqlalchemy_crud_plus/crud.py @@ -751,6 +751,7 @@ async def delete_model_by_column( allow_multiple: bool = False, logical_deletion: bool = False, deleted_flag_column: str = 'is_deleted', + deleted_flag_value: Any = True, deleted_at_column: str = 'deleted_at', deleted_at_factory: datetime | Callable[[], datetime] | None = None, flush: bool = False, @@ -764,6 +765,7 @@ async def delete_model_by_column( :param allow_multiple: If `True`, allows deleting multiple records that match the filters :param logical_deletion: If `True`, enable logical deletion instead of physical deletion :param deleted_flag_column: Column name for logical deletion flag + :param deleted_flag_value: Value written to logical deletion flag column :param deleted_at_column: Column name for delete time，automatic judgment :param deleted_at_factory: The delete time value or factory function :param flush: If `True`, flush all object changes to the database @@ -785,7 +787,7 @@ async def delete_model_by_column( if total_count > 1: raise MultipleResultsError(f'Only one record is expected to be deleted, found {total_count} records.') - data: dict[str, Any] = {deleted_flag_column: True} + data: dict[str, Any] = {deleted_flag_column: deleted_flag_value} if deleted_at_column in self.model_column_names: if deleted_at_factory is None: diff --git a/tests/models/basic.py b/tests/models/basic.py index 94d8c76..52f118a 100644 --- a/tests/models/basic.py +++ b/tests/models/basic.py @@ -16,8 +16,10 @@ class Ins(Base): id: Mapped[int] = mapped_column(init=False, primary_key=True, index=True, autoincrement=True) name: Mapped[str] = mapped_column(String(64)) is_deleted: Mapped[bool] = mapped_column(default=False) + deleted: Mapped[int] = mapped_column(default=0) created_time: Mapped[datetime] = mapped_column(init=False, default_factory=datetime.now) updated_time: Mapped[datetime | None] = mapped_column(init=False, onupdate=datetime.now) + deleted_time: Mapped[datetime | None] = mapped_column(init=False, default=None) class InsPks(Base): diff --git a/tests/test_delete.py b/tests/test_delete.py index 311e243..a0e9cdd 100644 --- a/tests/test_delete.py +++ b/tests/test_delete.py @@ -1,3 +1,5 @@ +from datetime import datetime + import pytest from sqlalchemy.ext.asyncio import AsyncSession @@ -148,6 +150,34 @@ async def test_logical_delete_single_record(db: AsyncSession, sample_ins: list[I assert updated_item.is_deleted is True +@pytest.mark.asyncio +async def test_logical_delete_with_deleted_flag_value_expression( + db: AsyncSession, sample_ins: list[Ins], crud_ins: CRUDPlus[Ins] +): + item = sample_ins[0] + + deleted_time = datetime(2024, 1, 1, 12, 0, 0) + + async with db.begin(): + count = await crud_ins.delete_model_by_column( + db, + logical_deletion=True, + deleted_flag_column='deleted', + deleted_flag_value=crud_ins.model.id, + deleted_at_column='deleted_time', + deleted_at_factory=deleted_time, + id=item.id, + deleted=0, + ) + + assert count == 1 + + updated_item = await crud_ins.select_model(db, item.id) + assert updated_item is not None + assert updated_item.deleted == item.id + assert updated_item.deleted_time == deleted_time + + @pytest.mark.asyncio async def test_logical_delete_multiple_records(db: AsyncSession, sample_ins: list[Ins], crud_ins: CRUDPlus[Ins]): before_count = await crud_ins.count(db, is_deleted=False)