Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

get_queryset filteration for related models #237

Open
nareshkmrtelidelhivery opened this issue Dec 10, 2023 · 0 comments
Open

get_queryset filteration for related models #237

nareshkmrtelidelhivery opened this issue Dec 10, 2023 · 0 comments

Comments

@nareshkmrtelidelhivery
Copy link

nareshkmrtelidelhivery commented Dec 10, 2023

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()

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

1 participant