makinacorpus / django-safedelete

Mask your objects instead of deleting them from your database.
https://django-safedelete.readthedocs.io/
BSD 3-Clause "New" or "Revised" License
677 stars 122 forks source link

get_queryset filteration for related models #237

Open nareshkmrtelidelhivery opened 10 months ago

nareshkmrtelidelhivery commented 10 months ago

When we override the get_queryset method it works for base models but not for related models is_deleated filter is not applied. here is the initial proposal for it [its not optimized code just an idea] please let me if there are any concerning points

from django.db import models
from django.db.models import sql
from django.db.models.query_utils import Q
from django.db.models.sql.compiler import SQLCompiler
from django.db.models.fields.related import RelatedField
from django.db.models.query import Collector

class ObjectStatus:
    DELETED = "deleted"
    ACTIVE = "active"

class Query(sql.Query):
    pass

class SoftDeleteQuerySet(models.query.QuerySet):
    def __init__(self, model=None, query=None, using=None, hints=None):
        super(SoftDeleteQuerySet, self).__init__(
            model=model, query=query, using=using, hints=hints
        )
        self.query: Query = query or Query(self.model)

    def delete(self):
        for obj in self.all():
            obj.delete()

    def filter(self, *args, **kwargs):
        queryset = self._clone()
        obj = super(SoftDeleteQuerySet, queryset).filter(*args, **kwargs)
        filters = obj.related_fields()
        print(filters)
        return obj._filter_or_exclude(False, args, filters)

    def related_field_soft_delete(
        self, model, visited: set, path: list, target_table: str
    ):
        if target_table == model._meta.db_table:
            if path:
                delete_filter_path = "__".join(["__".join(path), "status"])
            else:
                delete_filter_path = "status"
            return True, delete_filter_path

        visited.add(model)
        forward_fields = list(
            filter(lambda f: isinstance(f, RelatedField), model._meta.get_fields())
        )
        for field in forward_fields:
            _model, _on_delete, _name = (
                field.remote_field.model,
                field.remote_field.on_delete,
                field.name,
            )
            if not _model in visited:
                path.append(_name)
                is_table_found, delete_filter_path = self.related_field_soft_delete(
                    _model, visited, path, target_table
                )
                if is_table_found:
                    return True, delete_filter_path
                path.pop()

        reversed_fields = model._meta.related_objects
        for field in reversed_fields:
            _model, _on_delete, _name = (
                field.remote_field.model,
                field.remote_field.on_delete,
                field.name,
            )
            if not _model in visited:
                path.append(_name)
                is_table_found, delete_filter_path = self.related_field_soft_delete(
                    _model, visited, path, target_table
                )
                if is_table_found:
                    return True, delete_filter_path
                path.pop()
        return False, ""

    def related_fields(self):
        used_aliases = self.query.used_aliases
        table_map = self.query.table_map
        table_alias_map = dict()
        joined_tables = set()

        for table_name, aliases in table_map.items():
            for alias in aliases:
                table_alias_map[alias] = table_name
        for alias in used_aliases:
            joined_tables.add(table_alias_map[alias])

        filters = {}
        for target_table in joined_tables:
            target_table_path = []
            is_table_found, delete_filter_path = self.related_field_soft_delete(
                self.model, set(), target_table_path, target_table
            )
            if is_table_found:
                filters[delete_filter_path] = ObjectStatus.ACTIVE
        return filters

class BaseModelManager(models.Manager):
    def get_queryset(self):
        return SoftDeleteQuerySet(self.model, self._db)

class BaseModel(models.Model):
    status = models.CharField(max_length=100, default=ObjectStatus.ACTIVE)

    class Meta:
        abstract = True

class Author(BaseModel):
    name = models.CharField(max_length=100)
    objects = BaseModelManager()

class Children(BaseModel):
    name = models.CharField(max_length=100)
    author = models.ForeignKey(Author, on_delete=models.CASCADE)
    objects = BaseModelManager()

class Book(BaseModel):
    name = models.CharField(max_length=100)
    children = models.ForeignKey(
        Children, related_name="related_name_children", on_delete=models.CASCADE
    )
    objects = BaseModelManager()