Django中自关联Category与Product多对多的查询优化需求

Alright, let's tackle these two Django database optimization problems step by step—focused on cutting down query counts and ditching those inefficient Python-side recursive lookups. First, let's translate your existing database tables into standard Django models to set a clear foundation:

from django.db import models

class Category(models.Model):
    name = models.CharField(max_length=255)
    parent = models.ForeignKey('self', on_delete=models.CASCADE, null=True, blank=True, related_name='children')

    def __str__(self):
        return self.name

class Product(models.Model):
    name = models.CharField(max_length=255)
    categories = models.ManyToManyField(Category, related_name='products')

    def __str__(self):
        return self.name

1. Generate Breadcrumbs for a Product (First Category Path) #

The goal here is to grab the full parent hierarchy for the first category linked to a product, without chaining multiple parent-by-parent queries in Python. We'll use a recursive CTE (Common Table Expression) to precompute all category paths in one database hit—way more efficient than recursive Python loops.

Raw SQL Solution (Works with PostgreSQL/MySQL 8.0+) #

This is the most direct way to leverage database-level recursion:

from django.db import connection

def get_product_breadcrumbs(product_id):
    with connection.cursor() as cursor:
        cursor.execute("""
            WITH RECURSIVE category_paths AS (
                -- Start with top-level categories (no parent)
                SELECT id, name, parent_id, ARRAY[id] AS path_ids, ARRAY[name] AS path_names
                FROM category
                WHERE parent_id IS NULL
                UNION ALL
                -- Recursively add child categories to their parent's path
                SELECT c.id, c.name, c.parent_id, cp.path_ids || c.id, cp.path_names || c.name
                FROM category c
                JOIN category_paths cp ON c.parent_id = cp.id
            )
            -- Get the first category linked to the product, then its full path
            SELECT cp.path_ids, cp.path_names
            FROM product p
            JOIN product_categories pc ON p.id = pc.product_id
            JOIN category_paths cp ON pc.category_id = cp.id
            WHERE p.id = %s
            ORDER BY pc.id ASC  -- Ensures we pick the "first" category as per your intermediate table order
            LIMIT 1;
        """, [product_id])
        result = cursor.fetchone()
    
    if result:
        # Format the path to match your example: "Category Id 1 -> Cat id 3 -> Cat id 7"
        return " -> ".join([f"Cat id {id}" for id in result[0]])
    return None

ORM-Friendly Version (PostgreSQL, Django 3.2+) #

If you prefer sticking to Django's ORM (and use PostgreSQL), you can use the django_cte package to wrap the recursive logic:

from django.db.models import F, Value
from django.db.models.functions import ArrayAppend
from django.contrib.postgres.fields import ArrayField
from django_cte import CTEManager, CTEQuerySet

class CategoryCTEManager(CTEManager):
    def get_all_paths(self):
        # Build the recursive CTE to compute all category paths
        cte = self.with_cte(
            'category_paths',
            # Base case: top-level categories
            self.filter(parent__isnull=True).annotate(
                path_ids=ArrayField(Value([F('id')], output_field=ArrayField(models.IntegerField()))),
                path_names=ArrayField(Value([F('name')], output_field=ArrayField(models.CharField(max_length=255)))),
            ).union(
                # Recursive case: add children to parent paths
                self.filter(parent__isnull=False).annotate(
                    path_ids=ArrayAppend(F('parent__path_ids'), F('id')),
                    path_names=ArrayAppend(F('parent__path_names'), F('name')),
                ),
                all=True
            )
        )
        return cte.queryset()

class Category(models.Model):
    name = models.CharField(max_length=255)
    parent = models.ForeignKey('self', on_delete=models.CASCADE, null=True, blank=True, related_name='children')
    objects = CategoryCTEManager()

def get_product_breadcrumbs_orm(product_id):
    product = Product.objects.prefetch_related('categories').get(id=product_id)
    first_category = product.categories.first()
    if not first_category:
        return None
    
    # Fetch the precomputed path for the first category
    category_path = Category.objects.get_all_paths().get(id=first_category.id)
    return " -> ".join([f"Cat id {id}" for id in category_path.path_ids])

2. Get All Products from a Category & Its Subcategories #

Again, we'll use a recursive CTE to fetch every descendant category (including the original category) in one query, then join to products to avoid redundant lookups.

Raw SQL Solution #

from django.db import connection

def get_products_for_category(category_id):
    with connection.cursor() as cursor:
        cursor.execute("""
            WITH RECURSIVE category_tree AS (
                -- Start with the target category
                SELECT id FROM category WHERE id = %s
                UNION ALL
                -- Recursively add all child categories
                SELECT c.id FROM category c
                JOIN category_tree ct ON c.parent_id = ct.id
            )
            -- Get all products linked to any category in the tree (distinct to avoid duplicates)
            SELECT DISTINCT p.id, p.name
            FROM product p
            JOIN product_categories pc ON p.id = pc.product_id
            JOIN category_tree ct ON pc.category_id = ct.id;
        """, [category_id])
        results = cursor.fetchall()
    
    # Convert to Django Product instances if needed
    product_ids = [row[0] for row in results]
    return Product.objects.filter(id__in=product_ids).prefetch_related('categories')

ORM Version (PostgreSQL/MySQL 8.0+) #

Using django_cte to keep it clean:

from django_cte import CTEManager

class CategoryCTEManager(CTEManager):
    def get_descendants(self, category_id):
        cte = self.with_cte(
            'category_tree',
            # Base case: target category
            self.filter(id=category_id).values('id').union(
                # Recursive case: add all children
                self.filter(parent__in=CTEQuerySet('category_tree')).values('id'),
                all=True
            )
        )
        return cte.queryset().values_list('id', flat=True)

class Category(models.Model):
    name = models.CharField(max_length=255)
    parent = models.ForeignKey('self', on_delete=models.CASCADE, null=True, blank=True, related_name='children')
    objects = CategoryCTEManager()

def get_products_for_category_orm(category_id):
    # Get all descendant category IDs in one query
    descendant_ids = Category.objects.get_descendants(category_id)
    # Fetch distinct products, with prefetch to avoid N+1 queries later
    return Product.objects.filter(categories__id__in=descendant_ids).distinct().prefetch_related('categories')

Key Optimizations to Note: #

• Both solutions cut query counts from O(n) (where n is hierarchy depth) to 1-2 total queries by offloading recursion to the database, which is optimized for this kind of operation.
• Using DISTINCT ensures we don't return duplicate products that belong to multiple categories in the hierarchy.
• prefetch_related is added to avoid N+1 queries if you need to access product categories later.

内容的提问来源于stack exchange，提问作者user3541631