第5章：数据库操作

5.3 QuerySet和数据库API

QuerySet概念

QuerySet是Django ORM的核心概念，它表示数据库查询的集合。QuerySet具有以下特点：

惰性执行：QuerySet只有在需要结果时才会执行数据库查询
链式调用：可以连续调用多个方法构建复杂查询
可缓存：相同的QuerySet会缓存结果，避免重复查询

基本查询操作

获取所有对象

python

from myapp.models import Article

# 获取所有文章
articles = Article.objects.all()

# 转换为列表（触发查询执行）
article_list = list(articles)

# 遍历QuerySet（触发查询执行）
for article in articles:
    print(article.title)

获取单个对象

python

# 根据主键获取
article = Article.objects.get(pk=1)

# 根据字段获取
article = Article.objects.get(title="Django入门教程")

# 处理不存在的对象
try:
    article = Article.objects.get(pk=999)
except Article.DoesNotExist:
    print("文章不存在")

# 使用get_or_create
article, created = Article.objects.get_or_create(
    title="新文章",
    defaults={'content': '默认内容'}
)

过滤查询

python

# 基本过滤
published_articles = Article.objects.filter(is_published=True)

# 多条件过滤
recent_published = Article.objects.filter(
    is_published=True,
    created_at__gte=timezone.now() - timedelta(days=7)
)

# 排除查询
unpublished_articles = Article.objects.exclude(is_published=True)

# 链式过滤
articles = Article.objects.filter(category='tech').exclude(is_published=False)

字段查找（Field Lookups）

Django提供了丰富的字段查找方式：

python

# 精确匹配
articles = Article.objects.filter(title__exact="Django教程")

# 包含查询
articles = Article.objects.filter(title__contains="Django")

# 开头匹配  
articles = Article.objects.filter(title__startswith="Django")

# 结尾匹配
articles = Article.objects.filter(title__endswith="教程")

# 空值查询
articles = Article.objects.filter(content__isnull=True)

# 范围查询
from datetime import datetime, timedelta
start_date = datetime(2025, 1, 1)
end_date = datetime(2025, 12, 31)
articles = Article.objects.filter(created_at__range=(start_date, end_date))

# IN查询
categories = ['tech', 'science', 'art']
articles = Article.objects.filter(category__in=categories)

# 比较查询
articles = Article.objects.filter(views__gt=100)  # 大于
articles = Article.objects.filter(views__gte=50)  # 大于等于
articles = Article.objects.filter(views__lt=10)   # 小于
articles = Article.objects.filter(views__lte=5)   # 小于等于

排序和切片

python

# 升序排序
articles = Article.objects.order_by('created_at')

# 降序排序
articles = Article.objects.order_by('-created_at')

# 多字段排序
articles = Article.objects.order_by('category', '-created_at')

# 切片操作（LIMIT和OFFSET）
first_5_articles = Article.objects.all()[:5]          # LIMIT 5
next_5_articles = Article.objects.all()[5:10]         # LIMIT 5 OFFSET 5
last_5_articles = Article.objects.all()[-5:]          # 最后5条

# 随机排序
import random
articles = Article.objects.order_by('?')[:10]         # 随机10篇文章

聚合和注解

基本聚合

python

from django.db.models import Count, Avg, Max, Min, Sum

# 计数
article_count = Article.objects.count()
category_count = Article.objects.aggregate(Count('category'))

# 平均值
avg_views = Article.objects.aggregate(Avg('views'))

# 最大值和最小值
max_views = Article.objects.aggregate(Max('views'))
min_views = Article.objects.aggregate(Min('views'))

# 求和
total_views = Article.objects.aggregate(Sum('views'))

# 多聚合
stats = Article.objects.aggregate(
    total=Count('id'),
    avg_views=Avg('views'),
    max_views=Max('views')
)

分组聚合

python

# 按分类统计文章数量
category_stats = Article.objects.values('category').annotate(
    article_count=Count('id'),
    avg_views=Avg('views')
).order_by('-article_count')

# 按作者统计
author_stats = Article.objects.values('author__username').annotate(
    article_count=Count('id'),
    total_views=Sum('views')
)

# 按日期统计
from django.db.models.functions import TruncDate
daily_stats = Article.objects.annotate(
    date=TruncDate('created_at')
).values('date').annotate(
    count=Count('id')
).order_by('date')

注解（Annotations）

python

from django.db.models import F, Value
from django.db.models.functions import Concat, Coalesce

# 使用F表达式引用字段值
articles = Article.objects.annotate(
    popularity_score=F('views') * 0.7 + F('likes') * 0.3
)

# 字段计算
articles = Article.objects.annotate(
    reading_time=F('word_count') / 200  # 假设阅读速度200字/分钟
)

# 字符串拼接
articles = Article.objects.annotate(
    full_title=Concat('category', Value(': '), 'title')
)

# 处理空值
articles = Article.objects.annotate(
    safe_content=Coalesce('content', Value('暂无内容'))
)

# 组合使用多个函数
articles = Article.objects.annotate(
    display_title=Concat(
        Substr('title', 1, 20), 
        Value('...') if Length('title') > 20 else Value('')
    ),
    safe_summary=Coalesce('summary', Value('点击查看详情'))
)

# 使用F表达式进行字段运算
from django.db.models import F
articles = Article.objects.annotate(
    title_with_views=Concat('title', Value(' ('), F('view_count'), Value('次浏览)'))
)

关联查询

正向查询

python

# 一对一关系
class AuthorProfile(models.Model):
    author = models.OneToOneField(Author, on_delete=models.CASCADE)
    bio = models.TextField()

# 获取作者的个人简介
author = Author.objects.get(pk=1)
profile = author.authorprofile

# 一对多关系
class Comment(models.Model):
    article = models.ForeignKey(Article, on_delete=models.CASCADE)
    content = models.TextField()

# 获取文章的所有评论
article = Article.objects.get(pk=1)
comments = article.comment_set.all()

# 多对多关系
class Tag(models.Model):
    name = models.CharField(max_length=50)
    articles = models.ManyToManyField(Article)

# 获取文章的所有标签
article = Article.objects.get(pk=1)
tags = article.tag_set.all()

反向查询

python

# 通过外键反向查询
articles_with_comments = Article.objects.filter(comment__isnull=False)

# 通过多对多反向查询
articles_with_popular_tags = Article.objects.filter(
    tag__name__in=['django', 'python']
)

# 关联查询优化
articles = Article.objects.select_related('author').prefetch_related('tags')

查询优化

在模板中循环访问关联对象的属性时，一定要使用 select_related 或 prefetch_related 来预加载数据，这是Django性能优化的基本功，否则会产生N+1查询问题。

python

# select_related：用于一对一和一对多关系（SQL JOIN）
articles = Article.objects.select_related('author')  # 减少查询次数

# prefetch_related：用于多对多和反向关系（额外查询）
articles = Article.objects.prefetch_related('tags', 'comment_set')

# 组合使用,只产生1次查询
articles = Article.objects.select_related('author').prefetch_related(
    'tags', 'comment_set'
)

only 和 defer

python

# only：只加载指定字段
articles = Article.objects.only('title', 'created_at')

# defer：排除指定字段
articles = Article.objects.defer('content', 'metadata')

批量操作

python

# 批量创建
articles_data = [
    Article(title=f'文章{i}', content=f'内容{i}')
    for i in range(100)
]
Article.objects.bulk_create(articles_data)

# 批量更新
Article.objects.filter(is_published=False).update(is_published=True)

# 批量删除
Article.objects.filter(created_at__lt=timezone.now() - timedelta(days=365)).delete()

原生SQL查询

执行原生SQL

python

from django.db import connection

# 直接执行SQL
with connection.cursor() as cursor:
    cursor.execute("SELECT * FROM blog_article WHERE views > %s", [100])
    rows = cursor.fetchall()

# 使用raw方法
articles = Article.objects.raw('SELECT * FROM blog_article WHERE views > 100')

# 带参数的raw查询
articles = Article.objects.raw(
    'SELECT * FROM blog_article WHERE category = %s AND views > %s',
    ['tech', 50]
)

自定义SQL查询

python

# 自定义管理器方法
class ArticleManager(models.Manager):
    def popular_articles(self, min_views=100):
        return self.raw('''
            SELECT * FROM blog_article 
            WHERE views > %s 
            ORDER BY views DESC
        ''', [min_views])

class Article(models.Model):
    # ... 字段定义 ...
    objects = ArticleManager()

# 使用自定义查询
popular_articles = Article.objects.popular_articles(200)

实际示例：文章搜索功能

python

from django.db.models import Q

def search_articles(query, category=None, author=None):
    """
    文章搜索功能
    """
    # 构建查询条件
    conditions = Q()
    
    # 关键词搜索（标题或内容）
    if query:
        conditions |= Q(title__icontains=query)
        conditions |= Q(content__icontains=query)
    
    # 分类过滤
    if category:
        conditions &= Q(category=category)
    
    # 作者过滤
    if author:
        conditions &= Q(author__username=author)
    
    # 执行查询并优化
    articles = Article.objects.filter(conditions).select_related(
        'author'
    ).prefetch_related(
        'tags'
    ).order_by('-created_at')
    
    return articles

# 使用示例
results = search_articles('Django', category='tech')

高级查询技巧

条件表达式

python

from django.db.models import Case, When, Value, IntegerField

# 条件注解
articles = Article.objects.annotate(
    popularity=Case(
        When(views__gt=1000, then=Value('热门')),
        When(views__gt=100, then=Value('一般')),
        default=Value('冷门'),
        output_field=CharField()
    )
)

# 条件更新
Article.objects.update(
    status=Case(
        When(views__gt=1000, then=Value('popular')),
        default=Value('normal')
    )
)

窗口函数

窗口函数的优势

数据库端计算：减少应用层逻辑，提高性能
复杂分析：支持排名、分区、移动平均等高级分析
灵活性：可以组合多个窗口函数

使用注意事项

需要数据库支持窗口函数（PostgreSQL、MySQL 8.0+、SQLite 3.25+）
复杂的窗口函数可能影响查询性能
在模板中使用时要注意N+1查询问题

python

from django.db.models import Window, F, Avg, Sum
from django.db.models.functions import Rank,DenseRank,RowNumber,NthValue

# 按浏览量降序排名，获取前10名
# 结果特点：
# 相同浏览量的文章会获得相同的排名
# 排名会有间隔（如：1, 2, 2, 4, 5...）
articles = Article.objects.annotate(
    rank=Window(
        expression=Rank(),  # 使用Rank()排名函数
        order_by=F('views').desc()  # 按浏览量降序排列
    )
).filter(rank__lte=10)  # 只取排名前10的文章

# 密集排名（无间隔）
articles = Article.objects.annotate(
    dense_rank=Window(
        expression=DenseRank(),
        order_by=F('views').desc()
    )
)


# 为每篇文章添加行号（按创建时间升序）
articles = Article.objects.annotate(
    row_number=Window(
        expression=RowNumber(),  # 使用RowNumber()行号函数
        order_by=F('created_at').asc()  # 按创建时间升序排列
    )
)

# 分类内排名
articles = Article.objects.annotate(
    category_rank=Window(
        expression=Rank(),
        order_by=F('views').desc(),
        partition_by=F('category')  # 按分类分区
    )
)

# 移动平均（最近3篇文章的平均浏览量）
articles = Article.objects.annotate(
    moving_avg=Window(
        expression=Avg('views'),
        order_by=F('created_at').asc(),
        frame=Window.frame(start=-2, end=Window.CURRENT_ROW)
    )
)

性能监控和调试

查询分析

python

# 启用查询日志
import logging
logger = logging.getLogger('django.db.backends')
logger.setLevel(logging.DEBUG)

# 使用django-debug-toolbar
# 安装：pip install django-debug-toolbar
# 配置后可以查看详细的查询信息

# 手动检查查询
articles = Article.objects.all()
print(articles.query)  # 查看生成的SQL

查询优化建议

使用索引：为经常查询的字段创建数据库索引
避免N+1查询：使用select_related和prefetch_related
批量操作：使用bulk_create、bulk_update减少数据库往返
限制结果集：使用切片操作限制返回的数据量
缓存查询：对不经常变化的数据使用缓存

小结

Django的QuerySet和数据库API提供了强大而灵活的数据库操作能力：

✅ 丰富的查询方法：filter、exclude、get、annotate等
✅ 多种字段查找：精确匹配、范围查询、模糊搜索等
✅ 聚合和分组：Count、Sum、Avg等聚合函数
✅ 关联查询优化：select_related、prefetch_related
✅ 原生SQL支持：raw查询和自定义SQL
✅ 性能优化：批量操作、查询优化技巧

掌握这些功能可以帮助你构建高效、可维护的数据库查询逻辑。

我们将学习Django视图系统的使用。

6.1 函数视图 →

第5章：数据库操作

5.3 QuerySet和数据库API

QuerySet概念

基本查询操作

获取所有对象

获取单个对象

过滤查询

字段查找（Field Lookups）

排序和切片

聚合和注解

基本聚合

分组聚合

注解（Annotations）

关联查询

正向查询

反向查询

查询优化

only 和 defer

批量操作

原生SQL查询

执行原生SQL

自定义SQL查询

实际示例：文章搜索功能

高级查询技巧

条件表达式

窗口函数

性能监控和调试

查询分析

查询优化建议

小结

下一篇

目录

第5章：数据库操作 ​

5.3 QuerySet和数据库API ​

QuerySet概念 ​

基本查询操作 ​

获取所有对象 ​

获取单个对象 ​

过滤查询 ​

字段查找（Field Lookups） ​

排序和切片 ​

聚合和注解 ​

基本聚合 ​

分组聚合 ​

注解（Annotations） ​

关联查询 ​

正向查询 ​

反向查询 ​

查询优化 ​

select_related 和 prefetch_related ​

only 和 defer ​

批量操作 ​

原生SQL查询 ​

执行原生SQL ​

自定义SQL查询 ​

实际示例：文章搜索功能 ​

高级查询技巧 ​

条件表达式 ​

窗口函数 ​

性能监控和调试 ​

查询分析 ​

查询优化建议 ​

小结 ​

下一篇 ​

目录 ​

第5章：数据库操作

5.3 QuerySet和数据库API

QuerySet概念

基本查询操作

获取所有对象

获取单个对象

过滤查询

字段查找（Field Lookups）

排序和切片

聚合和注解

基本聚合

分组聚合

注解（Annotations）

关联查询

正向查询

反向查询

查询优化

select_related 和 prefetch_related

only 和 defer

批量操作

原生SQL查询

执行原生SQL

自定义SQL查询

实际示例：文章搜索功能

高级查询技巧

条件表达式

窗口函数

性能监控和调试

查询分析

查询优化建议

小结

下一篇

目录