Skip to content

Use Case 5: Soft Deletes Pattern

Scenario: Deleting a task should be recoverable. Instead of removing the document, set a tombstone flag and filter it from normal queries.

📦 What's New?

Component Description
SoftDeleteDocument Base model carrying is_deleted, deleted_at, deleted_by.
SoftDeleteRepository[T] Repository-level soft delete / restore / filtering using update + get_many.
permanent_delete Escapes to the real delete.

🚀 Example

Python
from datetime import datetime

from fastapi import FastAPI, HTTPException
from mongo_ops import BaseDocument, BaseRepository


class SoftDeleteDocument(BaseDocument):
    is_deleted: bool = False
    deleted_at: datetime | None = None
    deleted_by: str | None = None


class Task(SoftDeleteDocument):
    title: str = ""
    description: str = ""
    assignee_id: str = ""
    status: str = "pending"
    priority: str = "medium"


class SoftDeleteRepository(BaseRepository[T]):
    async def soft_delete(self, id: str, deleted_by: str | None = None):
        return await self.update(
            id,
            {"is_deleted": True, "deleted_at": datetime.utcnow(), "deleted_by": deleted_by},
        )

    async def restore(self, id: str):
        return await self.update(
            id, {"is_deleted": False, "deleted_at": None, "deleted_by": None}
        )

    async def get_active(self, skip: int = 0, limit: int = 100):
        return await self.get_many(filter={"is_deleted": False}, skip=skip, limit=limit)

    async def get_deleted(self, skip: int = 0, limit: int = 100):
        return await self.get_many(filter={"is_deleted": True}, skip=skip, limit=limit)

    async def permanent_delete(self, id: str) -> bool:
        return await self.delete(id)


class TaskRepository(SoftDeleteRepository[Task]):
    def __init__(self):
        super().__init__("tasks", Task)


app = FastAPI()
task_repo = TaskRepository()


@app.delete("/tasks/{task_id}")
async def soft_delete_task(task_id: str, user_id: str):
    task = await task_repo.soft_delete(task_id, deleted_by=user_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return {"message": "Task deleted", "task": task}


@app.post("/tasks/{task_id}/restore")
async def restore_task(task_id: str):
    task = await task_repo.restore(task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return {"message": "Task restored", "task": task}


@app.get("/tasks/", response_model=list[Task])
async def list_active_tasks(skip: int = 0, limit: int = 10):
    return await task_repo.get_active(skip, limit)


@app.get("/tasks/deleted", response_model=list[Task])
async def list_deleted_tasks(skip: int = 0, limit: int = 10):
    return await task_repo.get_deleted(skip, limit)

Note: add the lifespan wiring from use case 01 so task_repo is created after connection.

💡 Tips

  • BaseRepository[T] is generic — subclasses must parameterize it (see SoftDeleteRepository[T] above). from mongo_ops import BaseRepository; BaseRepository[T] works out of the box.
  • Keep a compound index on {"is_deleted": 1, "status": 1} for active-list queries (see use case 13).
  • Soft-deleted documents should be excluded at the repository boundary, never re-filtered ad hoc in endpoints.