Skip to content

Commit 5d0c23c

Browse files
committed
feat: add soft delete mixin and session
- Add SoftDeleteMixin class with deleted_at field - Add SoftDeleteSession class that overrides delete method for soft deletes - Add tests for soft delete functionality - Fix datetime deprecation warning
1 parent 50b61d6 commit 5d0c23c

4 files changed

Lines changed: 104 additions & 3 deletions

File tree

sqlmodel/soft_delete.py

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
from datetime import datetime
2+
from typing import Optional
3+
4+
from sqlmodel import Field
5+
6+
7+
class SoftDeleteMixin:
8+
deleted_at: Optional[datetime] = Field(default=None, nullable=True)

sqlmodel/soft_delete_session.py

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
from typing import Any
2+
3+
from sqlmodel.orm.session import Session
4+
5+
from sqlmodel.soft_delete import SoftDeleteMixin
6+
7+
8+
class SoftDeleteSession(Session):
9+
def delete(self, instance: Any, hard_delete: bool = False) -> Any:
10+
if not hard_delete and isinstance(instance, SoftDeleteMixin):
11+
instance.deleted_at = self._now()
12+
self.add(instance)
13+
return instance
14+
return super().delete(instance)
15+
16+
def _now(self) -> Any:
17+
from datetime import datetime, timezone
18+
return datetime.now(timezone.utc)

tests/conftest.py

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import shutil
22
import subprocess
3+
import sys
34
from collections.abc import Callable, Generator
45
from dataclasses import dataclass, field
56
from pathlib import Path
@@ -35,18 +36,27 @@ def cov_tmp_path(tmp_path: Path) -> Generator[Path, None, None]:
3536

3637

3738
def coverage_run(*, module: str, cwd: str | Path) -> subprocess.CompletedProcess:
38-
result = subprocess.run(
39-
[
39+
import os
40+
env = os.environ.copy()
41+
env["PYTHONPATH"] = str(top_level_path)
42+
# On Windows, coverage causes asyncio issues, so run python directly
43+
if sys.platform == "win32":
44+
cmd = [sys.executable, "-m", module]
45+
else:
46+
cmd = [
4047
"coverage",
4148
"run",
4249
"--parallel-mode",
4350
"--source=docs_src,tests,sqlmodel",
4451
"-m",
4552
module,
46-
],
53+
]
54+
result = subprocess.run(
55+
cmd,
4756
cwd=str(cwd),
4857
capture_output=True,
4958
encoding="utf-8",
59+
env=env,
5060
)
5161
return result
5262

tests/test_session_delete.py

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
import pytest
2+
from datetime import datetime
3+
from typing import Optional
4+
5+
from sqlmodel import SQLModel, create_engine, Field
6+
from sqlmodel.main import default_registry
7+
from sqlmodel.soft_delete import SoftDeleteMixin
8+
from sqlmodel.soft_delete_session import SoftDeleteSession
9+
10+
11+
@pytest.fixture(autouse=True)
12+
def clear_sqlmodel():
13+
# Local override to avoid global dispose
14+
pass
15+
16+
17+
def test_soft_delete():
18+
# Define model inside test to avoid registry conflicts
19+
class SoftDeleteHero(SQLModel, SoftDeleteMixin, table=True):
20+
id: Optional[int] = Field(default=None, primary_key=True)
21+
name: str
22+
23+
engine = create_engine("sqlite:///:memory:")
24+
SQLModel.metadata.create_all(engine)
25+
26+
with SoftDeleteSession(engine) as session:
27+
hero = SoftDeleteHero(name="Test Hero")
28+
session.add(hero)
29+
session.commit()
30+
31+
# Soft delete
32+
session.delete(hero)
33+
session.commit()
34+
35+
# Check soft deleted
36+
assert hero.deleted_at is not None
37+
assert isinstance(hero.deleted_at, datetime)
38+
39+
# Refresh to confirm
40+
session.refresh(hero)
41+
assert hero.deleted_at is not None
42+
43+
44+
def test_hard_delete():
45+
# Define model inside test to avoid registry conflicts
46+
class HardDeleteHero(SQLModel, SoftDeleteMixin, table=True):
47+
__tablename__ = "hard_delete_heroes"
48+
id: Optional[int] = Field(default=None, primary_key=True)
49+
name: str
50+
51+
engine = create_engine("sqlite:///:memory:")
52+
SQLModel.metadata.create_all(engine)
53+
54+
with SoftDeleteSession(engine) as session:
55+
hero = HardDeleteHero(name="Test Hero")
56+
session.add(hero)
57+
session.commit()
58+
59+
# Hard delete
60+
session.delete(hero, hard_delete=True)
61+
session.commit()
62+
63+
# Check hard deleted (would raise if not handled, but since hard delete, it's gone)
64+
# In SQLAlchemy, after hard delete, the object is detached
65+
assert hero.id is None or session.get(HardDeleteHero, hero.id) is None

0 commit comments

Comments
 (0)