django模型数据查询
模型定义示例
我们以一个博客系统为例,定义三个模型:Author、Category 和 Post。
from django.db import modelsclass Author(models.Model):name = models.CharField(max_length=100)email = models.EmailField()class Category(models.Model):name = models.CharField(max_length=50)class Post(models.Model):title = models.CharField(max_length=200)content = models.TextField()created_at = models.DateTimeField(auto_now_add=True)author = models.ForeignKey(Author, on_delete=models.CASCADE)categories = models.ManyToManyField(Category)is_published = models.BooleanField(default=False)
查询示例大全(基于上述模型)
1. 基本查询
# 获取所有文章
Post.objects.all()# 获取第一篇文章
Post.objects.first()# 获取最后一篇文章
Post.objects.last()# 获取主键为1的文章
Post.objects.get(pk=1)
2. 条件过滤(filter)
# 获取标题包含“Django”的文章
Post.objects.filter(title__icontains="Django")# 获取已发布的文章
Post.objects.filter(is_published=True)# 获取某个作者的所有文章
author = Author.objects.get(name="痘痘")
Post.objects.filter(author=author)
3. 排序(order_by)
# 按创建时间倒序排列
Post.objects.order_by('-created_at')# 按标题升序排列
Post.objects.order_by('title')
4. 限制数量(切片)
# 获取最新的5篇文章
Post.objects.order_by('-created_at')[:5]
5. 聚合与统计
from django.db.models import Count# 每个作者的文章数量
Author.objects.annotate(post_count=Count('post'))# 某个作者的文章总数
Post.objects.filter(author=author).count()
6. 多对多查询(ManyToMany)
# 获取某个分类下的所有文章
category = Category.objects.get(name="Python")
category.post_set.all()# 获取包含某个分类的文章
Post.objects.filter(categories=category)
7. 关联查询(select_related / prefetch_related)
# 查询文章并预加载作者信息(ForeignKey)
Post.objects.select_related('author').all()# 查询文章并预加载分类信息(ManyToMany)
Post.objects.prefetch_related('categories').all()
8. 组合查询(Q对象)
from django.db.models import Q# 获取标题包含“Django”或内容包含“ORM”的文章
Post.objects.filter(Q(title__icontains="Django") | Q(content__icontains="ORM"))
9. 排除查询(exclude)
# 获取未发布的文章
Post.objects.exclude(is_published=True)
10. 日期查询
from datetime import datetime# 获取2025年11月的文章
Post.objects.filter(created_at__year=2025, created_at__month=11)
如果你有特定的业务场景,比如“查询某作者在某分类下的文章”,我也可以帮你写出更复杂的组合查询。要不要我再举几个进阶例子?
还有吗?
当然还有,痘痘!下面是一些更进阶、更实战的 Django 查询技巧,适合你在企业级项目中提升性能、表达力和可维护性:
进阶查询技巧
11. exists() 与 values_list()
# 判断某作者是否有文章
Post.objects.filter(author=author).exists()# 获取所有文章的标题列表(返回 QuerySet)
Post.objects.values_list('title', flat=True)
12. values() 与字典查询结果
# 获取文章的标题和创建时间(返回字典列表)
Post.objects.values('title', 'created_at')
13. update() 与 delete()
# 批量更新:将所有未发布文章设为已发布
Post.objects.filter(is_published=False).update(is_published=True)# 删除某作者的所有文章
Post.objects.filter(author=author).delete()
14. F 表达式(字段间运算)
from django.db.models import F# 将文章标题改为“标题 + 作者名”
Post.objects.update(title=F('title') + ' by ' + F('author__name'))
15. Case/When 条件表达式(类似 SQL 的 CASE)
from django.db.models import Case, When, Value, CharField# 根据是否发布设置状态字段
Post.objects.annotate(status=Case(When(is_published=True, then=Value('已发布')),default=Value('草稿'),output_field=CharField())
)
16. Subquery 与 OuterRef(子查询)
from django.db.models import Subquery, OuterRef# 获取每篇文章的作者的最新文章标题
latest_post = Post.objects.filter(author=OuterRef('author')).order_by('-created_at')
Post.objects.annotate(latest_title=Subquery(latest_post.values('title')[:1]))
17. Raw SQL 查询(慎用)
Post.objects.raw("SELECT * FROM blog_post WHERE is_published = true")
18. 自定义 Manager 查询
# models.py
class PublishedPostManager(models.Manager):def get_queryset(self):return super().get_queryset().filter(is_published=True)class Post(models.Model):...objects = models.Manager() # 默认published = PublishedPostManager() # 自定义# 使用方式
Post.published.all()
19. 分页查询(Paginator)
from django.core.paginator import Paginatorposts = Post.objects.all()
paginator = Paginator(posts, 10) # 每页10条
page_1 = paginator.page(1)
20. 去重查询(distinct)
# 获取所有作者的邮箱(去重)
Author.objects.values('email').distinct()
高阶查询与架构技巧
21. 多层嵌套查询(链式 filter)
# 获取某作者发布的文章中,标题包含“Django”的
Post.objects.filter(author=author).filter(is_published=True).filter(title__icontains="Django")
22. 多字段模糊搜索(Q对象组合)
from django.db.models import Q# 搜索标题或内容包含关键字
keyword = "ORM"
Post.objects.filter(Q(title__icontains=keyword) | Q(content__icontains=keyword))
23. 动态字段查询(**kwargs)
# 动态构造查询条件
filters = {'is_published': True, 'author__name': '痘痘'}
Post.objects.filter(**filters)
24. 条件聚合(Count + filter)
from django.db.models import Count, Q# 每个作者发布文章的数量(只统计已发布)
Author.objects.annotate(published_count=Count('post', filter=Q(post__is_published=True)))
25. 使用数据库函数(Func、Lower、Length)
from django.db.models.functions import Lower, Length# 获取标题长度大于50的文章
Post.objects.annotate(title_length=Length('title')).filter(title_length__gt=50)# 按标题小写排序
Post.objects.order_by(Lower('title'))
26. 使用窗口函数(Window)
from django.db.models import Window
from django.db.models.functions import RowNumber# 给每篇文章加上作者内的序号
Post.objects.annotate(row_number=Window(expression=RowNumber(), partition_by=['author'], order_by=F('created_at').desc()))
27. 使用数据库视图(View)配合 ORM
# 在数据库中创建视图 view_published_posts
# 然后在 models.py 中映射
class PublishedPost(models.Model):title = models.CharField(max_length=200)...class Meta:managed = Falsedb_table = 'view_published_posts'
28. 使用数据库索引优化查询
class Post(models.Model):...class Meta:indexes = [models.Index(fields=['created_at']),models.Index(fields=['author', 'is_published']),]
29. 使用 .only() 和 .defer() 精简字段加载
# 只加载标题和作者,其他字段延迟加载
Post.objects.only('title', 'author')# 延迟加载 content 字段
Post.objects.defer('content')
30. 自定义 QuerySet 方法(链式调用)
# models.py
class PostQuerySet(models.QuerySet):def published(self):return self.filter(is_published=True)def by_author(self, author):return self.filter(author=author)class Post(models.Model):...objects = PostQuerySet.as_manager()# 使用方式
Post.objects.published().by_author(author)
超高阶 Django ORM 技巧
31. 使用 distinct() + annotate()(防止重复)
# 获取每个作者的最新文章标题(去重)
Post.objects.order_by('author', '-created_at').distinct('author')
注意:
distinct('field')只在 PostgreSQL 中支持。
32. 使用 union() 合并多个查询集
# 合并已发布和未发布的文章查询集
published = Post.objects.filter(is_published=True)
drafts = Post.objects.filter(is_published=False)
combined = published.union(drafts)
33. 使用 extra() 添加原生 SQL(已废弃但仍可用)
Post.objects.extra(select={'year': "EXTRACT(year FROM created_at)"})
推荐用
Func或RawSQL替代。
34. 使用 RawSQL 嵌入复杂 SQL 表达式
from django.db.models.expressions import RawSQLPost.objects.annotate(year=RawSQL("EXTRACT(year FROM created_at)", [])
)
35. 使用 Func 构造数据库函数表达式
from django.db.models import Func# 使用数据库函数 UPPER
Post.objects.annotate(upper_title=Func(F('title'), function='UPPER'))
36. 使用 Trunc 进行时间截断聚合
from django.db.models.functions import TruncMonth# 按月统计文章数量
Post.objects.annotate(month=TruncMonth('created_at')).values('month').annotate(count=Count('id'))
37. 使用 Coalesce 设置空值默认
from django.db.models.functions import Coalesce# 获取文章标题,如果为空则用“未命名”
Post.objects.annotate(display_title=Coalesce('title', Value('未命名')))
38. 使用 NullsFirst / NullsLast 控制排序空值位置(PostgreSQL)
from django.db.models import F
from django.db.models.functions import NullsLastPost.objects.order_by(NullsLast(F('created_at')))
39. 使用 FilteredRelation 做条件关联(Django 3.1+)
from django.db.models import FilteredRelation# 获取作者的已发布文章数量
Author.objects.annotate(published_posts=FilteredRelation('post', condition=Q(post__is_published=True))
).annotate(count=Count('published_posts'))
40. 使用 database router 做多数据库查询分发
# settings.py 中配置 router
DATABASE_ROUTERS = ['myapp.db_router.MyRouter']# router.py 中定义逻辑
class MyRouter:def db_for_read(self, model, **hints):if model._meta.app_label == 'analytics':return 'analytics_db'return 'default'
ORM + 架构实战技巧(终极篇)
41. select_for_update():事务中锁定行,防止并发写入冲突
from django.db import transactionwith transaction.atomic():post = Post.objects.select_for_update().get(pk=1)post.title = "更新标题"post.save()
✅ 用于并发场景下的乐观锁替代方案,确保数据一致性。
42. bulk_create() / bulk_update():批量插入/更新,极大提升性能
# 批量创建
Post.objects.bulk_create([Post(title="A", author=author),Post(title="B", author=author),
])# 批量更新
posts = Post.objects.filter(author=author)
for p in posts:p.is_published = True
Post.objects.bulk_update(posts, ['is_published'])
43. iterator():大数据量查询时节省内存
for post in Post.objects.all().iterator(chunk_size=1000):process(post)
44. chunked queryset + 分页处理大表
from django.core.paginator import Paginatorqs = Post.objects.all()
paginator = Paginator(qs, 1000)
for page_num in paginator.page_range:for post in paginator.page(page_num).object_list:process(post)
45. annotate(...).filter(...) vs .filter(...).annotate(...) 的顺序影响性能
# 更高效:先 filter 再 annotate,减少聚合数据量
Post.objects.filter(is_published=True).annotate(...)
46. 使用 QuerySet.explain() 分析 SQL 执行计划(PostgreSQL)
Post.objects.filter(is_published=True).explain()
47. 使用 database functions 构建复杂表达式(组合函数)
from django.db.models.functions import Concat, UpperPost.objects.annotate(display=Concat(Upper('title'), Value(' by '), F('author__name'))
)
48. 使用 Index + conditions 创建部分索引(PostgreSQL)
class Meta:indexes = [models.Index(fields=['created_at'], name='idx_recent_posts', condition=Q(is_published=True))]
49. 使用 signals 自动填充字段(如 slug、审计字段)
from django.db.models.signals import pre_save
from django.dispatch import receiver
from django.utils.text import slugify@receiver(pre_save, sender=Post)
def auto_slug(sender, instance, **kwargs):if not instance.slug:instance.slug = slugify(instance.title)
50. ORM 查询封装到 Repository 层(解耦业务逻辑)
# repositories/post_repository.py
class PostRepository:@staticmethoddef get_published_by_author(author):return Post.objects.filter(author=author, is_published=True)# views.py
posts = PostRepository.get_published_by_author(author)
实战场景下的 ORM 组合技
51. 条件字段更新(避免覆盖已有值)
# 只在 slug 为空时更新
Post.objects.filter(slug__isnull=True).update(slug=F('title'))
52. 使用 .aggregate() 做全局统计
from django.db.models import Avg, Max, Min# 所有文章的平均标题长度
Post.objects.aggregate(avg_len=Avg(Length('title')))
53. 使用 .count() vs .len() 的性能差异
# 推荐:数据库层统计
Post.objects.filter(is_published=True).count()# 不推荐:拉回 QuerySet 再计算
len(Post.objects.filter(is_published=True))
54. 使用 .exists() vs .count() > 0 的性能差异
# 推荐:exists() 更快
Post.objects.filter(author=author).exists()
55. 使用 .get_or_create() 简化查找+创建逻辑
author, created = Author.objects.get_or_create(name="痘痘", defaults={'email': 'dd@example.com'})
56. 使用 .update_or_create() 简化查找+更新逻辑
Post.objects.update_or_create(title="Django Tips",defaults={'content': 'Updated content', 'is_published': True}
)
57. 使用 .filter(...).delete() 替代 .get().delete()(防止异常)
# 推荐:不会抛异常
Post.objects.filter(pk=1).delete()
58. 使用 .filter(...).exists() 替代 .get() 判断(防止 DoesNotExist)
# 推荐:更安全
if Post.objects.filter(pk=1).exists():...
59. 使用 .filter(...).first() 替代 .get()(防止异常)
# 推荐:不会抛异常
post = Post.objects.filter(pk=1).first()
60. 使用 .select_related() + .only() 精准控制字段加载
Post.objects.select_related('author').only('title', 'author__name')
与其他模块协同的 ORM 实战
与 Django REST Framework(DRF)结合
-
在
get_queryset()中使用.select_related()/.prefetch_related()提升接口性能 -
使用
filter_backends+SearchFilter实现模糊搜索 -
使用
annotate()+ordering_fields实现动态排序字段
与 Celery 异步任务结合
-
ORM 查询结果作为任务参数传递(注意避免传 QuerySet,传 ID 或 dict)
-
在任务中使用
.select_for_update()保证并发安全
与权限系统结合(如 django-guardian)
-
查询某用户有权限的对象:
from guardian.shortcuts import get_objects_for_user get_objects_for_user(user, 'app.view_post')
与缓存结合(如 Redis)
-
查询结果缓存:
cache_key = f"post_{post_id}" post = cache.get(cache_key) if not post:post = Post.objects.get(pk=post_id)cache.set(cache_key, post, timeout=300)
