Nifty Django Feature: Model._meta.get_fields()

class Species(models.Model):
    name = models.CharField()

class Pet(models.Model):
    species = models.ForeignKey(Species, related_name="pet_set", on_delete=models.CASCADE)

class Vet(models.Model):
    specialty = models.ForeignKey(Species, related_name="vet_set", on_delete=models.CASCADE)

print(Species._meta.get_fields())
# (AutoField: my_app.species.id,
#  CharField: my_app.species.name,
#  <ManyToOneRel: my_app.pet>,
#  <ManyToOneRel: my_app.vet>)

class Species(models.Model):
    name = models.CharField()

class Pet(models.Model):
    species = models.ForeignKey(Species, on_delete=models.CASCADE)
    humans = models.ManyToManyField(User, blank=True)

class Nickname(models.Model):
    pet = models.ForeignKey(Pet, on_delete=models.CASCADE)

class Vet(models.Model):
    specialty = models.ForeignKey(Species, on_delete=models.CASCADE)

class Appointment(models.Model):
    pet = models.ForeignKey(Pet, on_delete=models.CASCADE)
    vet = models.ForeignKey(Vet, on_delete=models.CASCADE)

def delete_species(species):
    Appointment.objects.filter(pet__species=species).delete()
    Vet.objects.filter(specialty=species).delete()
    Nickname.objects.filter(pet__species=species).delete()
    Pet.objects.filter(species=species).delete()
    Species.objects.filter(id=species.id).delete()

def fields_referencing_model(model):
    """
    Find the all fields that reference the given model.

    For Species, return fields from the following Foreign Keys
      - Pet.species
        - Pet_human.pet (through model)
        - Appointment.pet
      - Vet.species
        - Appointment.vet
    """
    # include_hidden=True will include the relationships on
    # ManyToManyField(through=) models.
    for field in model._meta.get_fields(include_hidden=True):
        # Only do something with relationship fields
        # (ForeignKey and OneToOneField)
        if field.one_to_many or field.one_to_one:
            if not hasattr(field, "field"):
                # Skip ForeignKey/OneToOneField on this model,
                # we only want inbound relationships
                continue
            yield field
            # Recurse to support traveling the whole model graph
            yield from fields_referencing_model(field.related_model)

from graphlib import TopologicalSorter

def get_deletion_order(model):
    """Return all models that need to be deleted, before this model."""
    ts = TopologicalSorter()
    # Add the root.
    ts.add(model)
    for field in fields_referencing_model(model):
      # add(node, *predecessors)
        # Example: (Species, Vet)
        ts.add(field.model, field.related_model)
    # static_order() returns the nodes that had no predecessors,
    # working backwards until it reaches back to the root.
    return list(ts.static_order())

for i, m in enumerate(get_deletion_order(Species), 1):
    print(f"  {i}. {m.__name__}")
#  1. Pet_humans
#  2. Appointment
#  3. Nickname
#  4. Vet
#  5. Pet
#  6. Species

from django.apps import apps
from django.db import transaction
from celery import shared_task, chain


def schedule_delete(model_instance):
    """Schedule the deletion tasks to delete a given model incrementally."""
    root = model_instance._meta.model
    lookup_map = lookups_to_root(root)
    steps = [
        delete_task.si(
            model_label=model._meta.label,
            lookups=["pk"] if model is root else lookup_map[model],
            lookup_pk=model_instance.pk,
        )
        for model in get_deletion_order(root)
    ]
    return chain(*steps).apply_async()


@shared_task
def delete_task(model_label: str, lookups: list[str], lookup_pk: int):
    """
    Delete model instances that are related to the given lookups.
    """
    model = apps.get_model(model_label)
    with transaction.atomic():
        for lookup in lookups:
            model.objects.filter(**{lookup: lookup_pk}).delete()


def lookups_to_root(root):
    """Create list of paths from related models back to the root.

    Map each model to the relationship lookup(s) that walk back to a root
    instance, e.g.:
        Pet         -> ['species']
        Nickname    -> ['pet__species']
        Appointment -> ['pet__species', 'vet__specialty']
    """
    children = {}
    # Build a complete map of the relationships between models.
    for field in fields_referencing_model(root):
        children.setdefault(field.model, set()).add(
            (field.related_model, field.field.name)
        )

    paths = {}

    def _walk(model, lookup_so_far):
        for child, fk_name in children.get(model, ()):
            if lookup_so_far is None:
                child_lookup = fk_name
            else:
                child_lookup = f"{fk_name}__{lookup_so_far}"
            paths.setdefault(child, []).append(child_lookup)
            _walk(child, child_lookup)

    # Use the relationship map to generate the paths between
    # the root model and the related models.
    _walk(root, None)
    return paths