Django ORM – 多表实例（聚合与分组查询）

作者likuolei

以下是用中文详细讲解 Django ORM（对象关系映射）在多表操作中进行聚合与分组查询的指南，聚焦于使用聚合函数（Count、Sum、Avg 等）和分组（annotate、values）处理多表关联数据，涵盖一对多和多对多关系。我会保持简洁清晰，适合初学者和进阶用户，并提供实用示例，关键代码将包裹在 ““` 标签中以生成 artifact。如果你有特定问题或需要更深入的内容，请随时告知！

什么是 Django ORM 聚合与分组查询？

Django ORM 的聚合与分组查询用于对多表数据进行统计和分组分析。聚合函数（如 Count、Sum）计算汇总值，分组查询（通过 annotate 或 values）按字段分组，结合多表关联（如 ForeignKey、ManyToManyField）实现复杂数据分析。

  • 作用：
  • 统计关联数据（如每个分类的商品数量、总销售额）。
  • 按字段分组并计算汇总值（如按分类统计平均价格）。
  • 简化复杂 SQL（如 GROUP BY）的编写。
  • 特点：
  • 使用 django.db.models 的聚合函数和 annotate。
  • 支持跨表聚合和过滤。
  • 自动优化 SQL，防止注入风险。

1. 定义模型

假设我们有一个应用 myapp，包含三个模型：Category（分类）、Item（商品）、Tag（标签），展示一对多（Category:Item）和多对多（Item:Tag）关系。

示例：myapp/models.py

from django.db import models

class Category(models.Model):
    name = models.CharField(max_length=100, verbose_name='分类名称')
    created_at = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')

    def __str__(self):
        return self.name

    class Meta:
        db_table = 'categories'
        ordering = ['name']

class Item(models.Model):
    name = models.CharField(max_length=100, verbose_name='商品名称')
    price = models.DecimalField(max_digits=10, decimal_places=2, verbose_name='价格')
    description = models.TextField(blank=True, verbose_name='描述')
    is_active = models.BooleanField(default=True, verbose_name='是否上架')
    category = models.ForeignKey(
        Category,
        on_delete=models.CASCADE,
        related_name='items',
        verbose_name='分类'
    )
    tags = models.ManyToManyField('Tag', related_name='items', verbose_name='标签')

    def __str__(self):
        return self.name

    class Meta:
        db_table = 'items'
        ordering = ['-price']

class Tag(models.Model):
    name = models.CharField(max_length=50, verbose_name='标签名称')

    def __str__(self):
        return self.name

    class Meta:
        db_table = 'tags'
        ordering = ['name']


from django.db import models

class Category(models.Model):
name = models.CharField(max_length=100, verbose_name=’分类名称’)
created_at = models.DateTimeField(auto_now_add=True, verbose_name=’创建时间’)

def __str__(self):
    return self.name

class Meta:
    db_table = 'categories'
    ordering = ['name']

class Item(models.Model):
name = models.CharField(max_length=100, verbose_name=’商品名称’)
price = models.DecimalField(max_digits=10, decimal_places=2, verbose_name=’价格’)
description = models.TextField(blank=True, verbose_name=’描述’)
is_active = models.BooleanField(default=True, verbose_name=’是否上架’)
category = models.ForeignKey(
Category,
on_delete=models.CASCADE,
related_name=’items’,
verbose_name=’分类’
)
tags = models.ManyToManyField(‘Tag’, related_name=’items’, verbose_name=’标签’)

def __str__(self):
    return self.name

class Meta:
    db_table = 'items'
    ordering = ['-price']

class Tag(models.Model):
name = models.CharField(max_length=50, verbose_name=’标签名称’)

def __str__(self):
    return self.name

class Meta:
    db_table = 'tags'
    ordering = ['name']

“`python
from django.db import models

class Category(models.Model):
name = models.CharField(max_length=100, verbose_name=’分类名称’)
created_at = models.DateTimeField(auto_now_add=True, verbose_name=’创建时间’)

def __str__(self):
    return self.name

class Meta:
    db_table = 'categories'
    ordering = ['name']

class Item(models.Model):
name = models.CharField(max_length=100, verbose_name=’商品名称’)
price = models.DecimalField(max_digits=10, decimal_places=2, verbose_name=’价格’)
description = models.TextField(blank=True, verbose_name=’描述’)
is_active = models.BooleanField(default=True, verbose_name=’是否上架’)
category = models.ForeignKey(
Category,
on_delete=models.CASCADE,
related_name=’items’,
verbose_name=’分类’
)
tags = models.ManyToManyField(‘Tag’, related_name=’items’, verbose_name=’标签’)

def __str__(self):
    return self.name

class Meta:
    db_table = 'items'
    ordering = ['-price']

class Tag(models.Model):
name = models.CharField(max_length=50, verbose_name=’标签名称’)

def __str__(self):
    return self.name

class Meta:
    db_table = 'tags'
    ordering = ['name']
- **说明**：
  - `Category:Item`：一对多（一个分类多个商品）。
  - `Item:Tag`：多对多（一个商品多个标签，一个标签多个商品）。
  - `related_name`：支持反向查询（如 `category.items`）。
  - 运行迁移：
    ```bash
    python manage.py makemigrations
    python manage.py migrate
    ```

---

### 2. 聚合与分组查询
以下展示如何对多表数据进行聚合和分组查询。

#### 2.1 聚合查询
使用 `aggregate()` 计算整个查询集的汇总值。

##### 示例：统计所有商品

python
from django.shortcuts import render
from django.db.models import Count, Sum, Avg, Max, Min
from .models import Item

def stats_view(request):
# 统计所有商品数量、总价、平均价、最高价、最低价
stats = Item.objects.aggregate(
total_count=Count(‘id’),
total_price=Sum(‘price’),
avg_price=Avg(‘price’),
max_price=Max(‘price’),
min_price=Min(‘price’)
)

return render(request, 'myapp/stats.html', {
    'stats': stats,
})
- **说明**：
  - `aggregate()`：返回字典，包含聚合结果。
  - 结果示例：`{'total_count': 100, 'total_price': Decimal('5999.00'), ...}`。
  - 过滤后聚合：
    ```python
    active_stats = Item.objects.filter(is_active=True).aggregate(total_count=Count('id'))
    ```

#### 2.2 分组查询
使用 `annotate()` 为每个对象添加聚合值，或使用 `values()` 按字段分组。

##### 示例：按分类分组统计

python
from django.shortcuts import render
from django.db.models import Count, Sum, Avg
from .models import Category, Item, Tag

def category_stats(request):
# 每个分类的商品数量和平均价格
categories = Category.objects.annotate(
item_count=Count(‘items’),
avg_price=Avg(‘items__price’)
)

# 按分类分组统计总销售额
sales_by_category = Item.objects.values('category__name').annotate(
    total_sales=Sum('price')
).order_by('-total_sales')

# 每个标签的商品数量
tags = Tag.objects.annotate(
    item_count=Count('items')
)

return render(request, 'myapp/category_stats.html', {
    'categories': categories,
    'sales_by_category': sales_by_category,
    'tags': tags,
})
- **说明**：
  - `annotate()`：为每个 `Category` 或 `Tag` 对象添加计算字段（如 `item_count`）。
  - `values()`：按指定字段分组，返回字典列表。
  - 跨表聚合：使用 `__` 访问关联字段（如 `items__price`）。
  - 结果示例：
    ```python
    # categories: [<Category: 书籍>, item_count=10, avg_price=59.99]
    # sales_by_category: [{'category__name': '书籍', 'total_sales': 599.90}]
    ```

#### 2.3 复杂查询
结合过滤、聚合和分组。

##### 示例：筛选后分组

python
from django.shortcuts import render
from django.db.models import Count, Sum
from .models import Item

def complex_stats(request):
# 仅统计活跃商品，按分类和标签分组
active_items = Item.objects.filter(is_active=True).values(
‘category__name’, ‘tags__name’
).annotate(
item_count=Count(‘id’),
total_price=Sum(‘price’)
).order_by(‘category__name’, ‘-item_count’)

return render(request, 'myapp/complex_stats.html', {
    'stats': active_items,
})
- **说明**：
  - 过滤：`is_active=True`。
  - 分组：按 `category__name` 和 `tags__name`。
  - 结果示例：
    ```python
    [{'category__name': '书籍', 'tags__name': 'Python', 'item_count': 5, 'total_price': 299.95}, ...]
    ```

---

### 3. 模板和视图集成
以下是将聚合与分组查询结果展示到前端的示例。

#### 示例：模板 `myapp/templates/myapp/stats.html`

html
{% extends ‘base.html’ %}

{% block content %}

商品统计

总商品数：{{ stats.total_count }}

总价格：￥{{ stats.total_price|floatformat:2 }}

平均价格：￥{{ stats.avg_price|floatformat:2 }}

最高价格：￥{{ stats.max_price|floatformat:2 }}

最低价格：￥{{ stats.min_price|floatformat:2 }}
{% endblock %}

#### 示例：模板 `myapp/templates/myapp/category_stats.html`

html
{% extends ‘base.html’ %}

{% block content %}

分类统计

分类详情

  • {{ category.name }} – 商品数: {{ category.item_count }} – 平均价格: ￥{{ category.avg_price|floatformat:2 }}
  • 暂无分类
<h2>分类销售额</h2>
<ul>
{% for sale in sales_by_category %}
    <li>{{ sale.category__name }} - 总销售额: ￥{{ sale.total_sales|floatformat:2 }}</li>
{% empty %}
    <li>暂无销售额数据</li>
{% endfor %}
</ul>

<h2>标签统计</h2>
<ul>
{% for tag in tags %}
    <li>{{ tag.name }} - 商品数: {{ tag.item_count }}</li>
{% empty %}
    <li>暂无标签</li>
{% endfor %}
</ul>

{% endblock %}

#### 示例：模板 `myapp/templates/myapp/complex_stats.html`

html
{% extends ‘base.html’ %}

{% block content %}

复杂统计

  • 分类: {{ stat.category__name }} – 标签: {{ stat.tags__name|default:”无标签” }} – 商品数: {{ stat.item_count }} – 总价格: ￥{{ stat.total_price|floatformat:2 }}
  • 暂无数据

{% endblock %}

#### 示例：基础模板 `templates/base.html`

html

{% block title %}我的网站{% endblock %} 统计 | 分类统计 | 复杂统计 {% block content %} {% endblock %} © 2025 我的网站

---

### 4. URL 路由
将视图绑定到 URL。

#### 示例：`myapp/urls.py`

python
from django.urls import path
from . import views

app_name = ‘myapp’

urlpatterns = [
path(‘stats/’, views.stats_view, name=’stats’),
path(‘category-stats/’, views.category_stats, name=’category_stats’),
path(‘complex-stats/’, views.complex_stats, name=’complex_stats’),
]

#### 项目级：`myproject/urls.py`

python
from django.contrib import admin
from django.urls import path, include

urlpatterns = [
path(‘admin/’, admin.site.urls),
path(”, include(‘myapp.urls’, namespace=’myapp’)),
]

---

### 5. 数据准备
为测试聚合查询，插入示例数据。

#### 示例：`myapp/views.py`（初始化数据）

python
from django.shortcuts import render
from .models import Category, Item, Tag

def init_data(request):
# 清空数据（开发环境）
Item.objects.all().delete()
Category.objects.all().delete()
Tag.objects.all().delete()

# 创建分类
books = Category.objects.create(name='书籍')
electronics = Category.objects.create(name='电子产品')

# 创建标签
python = Tag.objects.create(name='Python')
django = Tag.objects.create(name='Django')
tech = Tag.objects.create(name='技术')

# 创建商品
item1 = Item.objects.create(name='Python 编程', price=59.99, category=books)
item1.tags.add(python, tech)
item2 = Item.objects.create(name='Django 开发', price=79.99, category=books)
item2.tags.add(django, tech)
item3 = Item.objects.create(name='笔记本电脑', price=4999.99, category=electronics)
item3.tags.add(tech)

return render(request, 'myapp/success.html', {'message': '数据初始化成功'})
- **添加路由**：

python
path(‘init/’, views.init_data, name=’init_data’),

---

### 6. 最佳实践
1. **查询优化**：
   - 使用 `select_related` 和 `prefetch_related`：
     ```python
     categories = Category.objects.select_related().annotate(item_count=Count('items'))
     ```
   - 避免 N+1 问题：
     ```python
     # 差：for category in Category.objects.all(): print(category.items.count())
     # 好：categories = Category.objects.annotate(item_count=Count('items'))
     ```

2. **明确字段**：
   - 使用 `values()` 减少返回数据：
     ```python
     sales = Item.objects.values('category__name').annotate(total=Sum('price'))
     ```

3. **异常处理**：
   - 检查查询结果是否为空：
     ```python
     stats = Item.objects.aggregate(total_count=Count('id'))
     if not stats['total_count']:
         return render(request, 'myapp/error.html', {'message': '无数据'})
     ```

4. **安全性**：
   - 使用 ORM 聚合函数，避免原生 SQL。
   - 验证用户输入（如通过 `form.cleaned_data`）。

5. **调试查询**：
   - 查看 SQL：
     ```python
     print(Category.objects.annotate(item_count=Count('items')).query)
     ```
   - 使用 Django Debug Toolbar 分析性能。

---

### 7. 常见问题
1. **聚合结果为空**：
   - 检查数据库是否有数据（`Item.objects.count()`）。
   - 确认过滤条件是否过严。

2. **分组查询重复**：
   - 使用 `distinct()` 避免重复：
     ```python
     items = Item.objects.values('category__name').distinct()
     ```

3. **性能慢**：
   - 确保使用 `select_related` 或 `prefetch_related`。
   - 限制 `values()` 字段：
     ```python
     Item.objects.values('category__name').annotate(Count('id'))
     ```

4. **多对多分组错误**：
   - 正确使用 `prefetch_related`：
     ```python
     tags = Tag.objects.prefetch_related('items').annotate(item_count=Count('items'))
     ```

---

### 8. 运行项目
1. 确保模型迁移：

bash
python manage.py makemigrations
python manage.py migrate

2. 启动服务器：

bash
python manage.py runserver
“`

  1. 初始化数据：
  • 访问：http://127.0.0.1:8000/init/
  1. 访问统计页面：
  • 总体统计：http://127.0.0.1:8000/stats/
  • 分类统计：http://127.0.0.1:8000/category-stats/
  • 复杂统计：http://127.0.0.1:8000/complex-stats/

总结

Django ORM 的聚合与分组查询通过 aggregate()、annotate() 和 values() 实现多表数据分析。Count、Sum、Avg 等函数支持汇总统计，跨表查询和优化工具（如 select_related、prefetch_related）确保高效执行。结合模板展示结果，遵循最佳实践（如优化查询、异常处理）可构建强大的数据分析功能。

如果你需要更复杂的 ORM 示例（如子查询、条件聚合）、特定功能的深入讲解，或调试聚合查询问题的方法，请告诉我！

类似文章

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注