Django 模型与 ORM

一、模型基础

1.1 什么是 Django 模型

Django 模型是与数据库交互的核心组件，每个模型类对应数据库中的一张表。通过模型，开发者可以使用 Python 代码定义数据结构，而无需编写 SQL 语句。

from django.db import models

class Article(models.Model):
    """文章模型"""
    title = models.CharField('标题', max_length=200)
    content = models.TextField('内容')
    created_at = models.DateTimeField('创建时间', auto_now_add=True)

    class Meta:
        verbose_name = '文章'
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.title

1.2 字段类型详解

Django 提供了丰富的字段类型：

字符串字段

class Profile(models.Model):
    # 定长字符串，必须指定 max_length
    username = models.CharField('用户名', max_length=50)

    # 长文本，无长度限制
    bio = models.TextField('个人简介', blank=True)

    # Slug 字段，用于 URL
    slug = models.SlugField('URL别名', unique=True)

    # Email 字段，自带验证
    email = models.EmailField('邮箱')

    # URL 字段
    website = models.URLField('网站', blank=True)

    # UUID 字段
    uuid = models.UUIDField('UUID', default=uuid.uuid4)

    # IP 地址字段
    ip_address = models.GenericIPAddressField('IP地址', null=True)

数字字段

class Product(models.Model):
    # 整数字段
    stock = models.IntegerField('库存', default=0)

    # 正整数
    views = models.PositiveIntegerField('浏览量', default=0)

    # 小整数（-32768 到 32767）
    rating = models.SmallIntegerField('评分')

    # 大整数
    total_sales = models.BigIntegerField('总销量', default=0)

    # 浮点数
    weight = models.FloatField('重量')

    # 精确小数，适合金额
    price = models.DecimalField('价格', max_digits=10, decimal_places=2)

日期时间字段

class Event(models.Model):
    # 日期字段
    date = models.DateField('日期')

    # 时间字段
    time = models.TimeField('时间')

    # 日期时间字段
    datetime = models.DateTimeField('日期时间')

    # 自动设置为创建时间
    created_at = models.DateTimeField('创建时间', auto_now_add=True)

    # 自动更新为修改时间
    updated_at = models.DateTimeField('更新时间', auto_now=True)

    # 时间间隔
    duration = models.DurationField('持续时间')

布尔和选择字段

class Task(models.Model):
    STATUS_CHOICES = [
        ('pending', '待处理'),
        ('processing', '处理中'),
        ('completed', '已完成'),
        ('cancelled', '已取消'),
    ]

    # 布尔字段
    is_active = models.BooleanField('是否激活', default=True)

    # 可为空的布尔字段（三态）
    is_verified = models.BooleanField('是否验证', null=True)

    # 选择字段
    status = models.CharField(
        '状态',
        max_length=20,
        choices=STATUS_CHOICES,
        default='pending'
    )

    # 整数选择
    PRIORITY_CHOICES = [(1, '低'), (2, '中'), (3, '高')]
    priority = models.IntegerField('优先级', choices=PRIORITY_CHOICES, default=2)

文件字段

import os
from django.utils import timezone

def upload_to_path(instance, filename):
    """自定义上传路径"""
    ext = filename.split('.')[-1]
    date = timezone.now().strftime('%Y/%m/%d')
    return f'uploads/{date}/{instance.id}.{ext}'

class Document(models.Model):
    # 文件字段
    file = models.FileField('文件', upload_to='documents/')

    # 自定义上传路径
    attachment = models.FileField('附件', upload_to=upload_to_path)

    # 图片字段（自动验证图片格式）
    image = models.ImageField('图片', upload_to='images/')

    # 带尺寸的图片字段
    photo = models.ImageField(
        '照片',
        upload_to='photos/',
        width_field='photo_width',
        height_field='photo_height'
    )
    photo_width = models.IntegerField('照片宽度', null=True)
    photo_height = models.IntegerField('照片高度', null=True)

JSON 字段

class Configuration(models.Model):
    # JSON 字段（Django 3.1+）
    settings = models.JSONField('设置', default=dict)

    # 带默认值
    metadata = models.JSONField('元数据', default=list, blank=True)

1.3 字段选项详解

class Article(models.Model):
    title = models.CharField(
        '标题',                    # verbose_name: 字段的可读名称
        max_length=200,            # 最大长度
        unique=True,               # 唯一约束
        db_index=True,             # 创建索引
        help_text='文章的标题',     # 帮助文本
    )

    slug = models.SlugField(
        'URL别名',
        unique=True,
        db_column='url_slug',      # 自定义数据库列名
    )

    content = models.TextField(
        '内容',
        blank=True,                # 允许表单为空
        null=True,                 # 允许数据库为 NULL
        default='',                # 默认值
    )

    author = models.ForeignKey(
        'auth.User',
        on_delete=models.CASCADE,
        related_name='articles',   # 反向关系名称
        related_query_name='article',  # 查询时的名称
        verbose_name='作者',
    )

    status = models.CharField(
        '状态',
        max_length=20,
        choices=[('draft', '草稿'), ('published', '已发布')],
        default='draft',
        editable=True,             # 是否可编辑
    )

    created_at = models.DateTimeField(
        '创建时间',
        auto_now_add=True,         # 创建时自动设置
        editable=False,            # Admin 中不可编辑
    )

    # 数据库注释（Django 4.2+）
    description = models.TextField(
        '描述',
        db_comment='文章的详细描述',
    )

1.4 Meta 选项

class Article(models.Model):
    title = models.CharField(max_length=200)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        # 数据库表名
        db_table = 'blog_articles'

        # 可读名称
        verbose_name = '文章'
        verbose_name_plural = '文章列表'

        # 默认排序
        ordering = ['-created_at', 'title']

        # 获取最新记录的字段
        get_latest_by = 'created_at'

        # 唯一性约束
        unique_together = [['author', 'slug']]

        # 索引
        indexes = [
            models.Index(fields=['title'], name='article_title_idx'),
            models.Index(fields=['status', '-created_at']),
            models.Index(
                fields=['title'],
                condition=models.Q(status='published'),
                name='published_article_idx'
            ),
        ]

        # 约束
        constraints = [
            models.CheckConstraint(
                check=models.Q(views__gte=0),
                name='views_non_negative'
            ),
            models.UniqueConstraint(
                fields=['author', 'title'],
                name='unique_author_title'
            ),
        ]

        # 权限
        permissions = [
            ('can_publish', '可以发布文章'),
            ('can_feature', '可以推荐文章'),
        ]

        # 抽象模型
        abstract = False

        # 代理模型
        proxy = False

二、关联关系

2.1 一对多关系（ForeignKey）

class Category(models.Model):
    name = models.CharField('分类名', max_length=100)

class Article(models.Model):
    title = models.CharField('标题', max_length=200)
    category = models.ForeignKey(
        Category,
        on_delete=models.CASCADE,      # 删除行为
        related_name='articles',        # 反向查询名称
        related_query_name='article',   # 查询过滤名称
        verbose_name='分类',
        null=True,                      # 允许为空
        blank=True,
        limit_choices_to={'is_active': True},  # 限制选择
        db_constraint=True,             # 是否创建外键约束
    )

# on_delete 选项：
# CASCADE: 级联删除
# PROTECT: 阻止删除（抛出 ProtectedError）
# SET_NULL: 设为 NULL（需要 null=True）
# SET_DEFAULT: 设为默认值（需要 default）
# SET(): 设为指定值或调用函数
# DO_NOTHING: 不做任何操作（可能导致完整性错误）

# 使用示例
category = Category.objects.get(pk=1)
articles = category.articles.all()  # 反向查询

article = Article.objects.get(pk=1)
category = article.category  # 正向查询

# 查询过滤
Category.objects.filter(article__title='测试')  # related_query_name

2.2 一对一关系（OneToOneField）

from django.contrib.auth.models import User

class UserProfile(models.Model):
    user = models.OneToOneField(
        User,
        on_delete=models.CASCADE,
        related_name='profile',
        primary_key=True,  # 作为主键
    )
    avatar = models.ImageField('头像', upload_to='avatars/')
    bio = models.TextField('简介', blank=True)

    class Meta:
        verbose_name = '用户资料'

# 使用示例
user = User.objects.get(pk=1)
profile = user.profile  # 访问关联的 profile

profile = UserProfile.objects.get(pk=1)
user = profile.user  # 访问关联的 user

2.3 多对多关系（ManyToManyField）

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

class Article(models.Model):
    title = models.CharField('标题', max_length=200)
    tags = models.ManyToManyField(
        Tag,
        related_name='articles',
        blank=True,
        verbose_name='标签',
    )

# 使用示例
article = Article.objects.get(pk=1)
tags = article.tags.all()  # 获取所有标签

# 添加标签
tag = Tag.objects.get(pk=1)
article.tags.add(tag)
article.tags.add(1, 2, 3)  # 通过 ID 添加

# 移除标签
article.tags.remove(tag)

# 清空标签
article.tags.clear()

# 设置标签（替换现有的）
article.tags.set([1, 2, 3])

# 反向查询
tag = Tag.objects.get(pk=1)
articles = tag.articles.all()

2.4 自定义中间表

class Article(models.Model):
    title = models.CharField('标题', max_length=200)
    tags = models.ManyToManyField(
        'Tag',
        through='ArticleTag',
        through_fields=('article', 'tag'),
    )

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

class ArticleTag(models.Model):
    """文章标签关联表"""
    article = models.ForeignKey(Article, on_delete=models.CASCADE)
    tag = models.ForeignKey(Tag, on_delete=models.CASCADE)
    created_at = models.DateTimeField('添加时间', auto_now_add=True)
    created_by = models.ForeignKey(
        'auth.User',
        on_delete=models.SET_NULL,
        null=True,
    )

    class Meta:
        unique_together = ['article', 'tag']
        ordering = ['-created_at']

# 使用中间表
ArticleTag.objects.create(
    article=article,
    tag=tag,
    created_by=user
)

2.5 自关联

class Category(models.Model):
    """支持无限层级的分类"""
    name = models.CharField('分类名', max_length=100)
    parent = models.ForeignKey(
        'self',
        on_delete=models.CASCADE,
        null=True,
        blank=True,
        related_name='children',
    )

    def get_ancestors(self):
        """获取所有祖先分类"""
        ancestors = []
        current = self.parent
        while current:
            ancestors.append(current)
            current = current.parent
        return ancestors

    def get_descendants(self):
        """获取所有后代分类"""
        descendants = list(self.children.all())
        for child in self.children.all():
            descendants.extend(child.get_descendants())
        return descendants

class Comment(models.Model):
    """评论回复"""
    content = models.TextField('内容')
    parent = models.ForeignKey(
        'self',
        on_delete=models.CASCADE,
        null=True,
        blank=True,
        related_name='replies',
    )

三、QuerySet API

3.1 创建对象

# 方式一：实例化后保存
article = Article(title='标题', content='内容')
article.save()

# 方式二：create 方法（推荐）
article = Article.objects.create(
    title='标题',
    content='内容',
    author=user,
)

# 方式三：get_or_create（获取或创建）
article, created = Article.objects.get_or_create(
    slug='my-article',
    defaults={
        'title': '我的文章',
        'content': '内容',
        'author': user,
    }
)

# 方式四：update_or_create（更新或创建）
article, created = Article.objects.update_or_create(
    slug='my-article',
    defaults={
        'title': '更新的标题',
        'content': '更新的内容',
    }
)

# 方式五：批量创建
articles = Article.objects.bulk_create([
    Article(title='文章1', content='内容1'),
    Article(title='文章2', content='内容2'),
    Article(title='文章3', content='内容3'),
], batch_size=100)

# 批量创建时忽略冲突
Article.objects.bulk_create(
    articles,
    ignore_conflicts=True,
    update_conflicts=True,
    update_fields=['title', 'content'],
    unique_fields=['slug'],
)

3.2 查询方法

# 获取所有对象
articles = Article.objects.all()

# 获取单个对象
article = Article.objects.get(pk=1)  # 不存在会抛出异常
article = Article.objects.filter(pk=1).first()  # 不存在返回 None

# 过滤查询
articles = Article.objects.filter(status='published')
articles = Article.objects.exclude(status='draft')

# 链式查询
articles = Article.objects.filter(
    status='published'
).exclude(
    author=user
).order_by('-created_at')

# 切片（LIMIT 和 OFFSET）
articles = Article.objects.all()[:10]  # LIMIT 10
articles = Article.objects.all()[5:10]  # OFFSET 5 LIMIT 5

# 排序
articles = Article.objects.order_by('-created_at', 'title')
articles = Article.objects.order_by('?')  # 随机排序

# 去重
articles = Article.objects.distinct()
articles = Article.objects.values('category').distinct()

# 反转
articles = Article.objects.order_by('created_at').reverse()

3.3 字段查找

# 精确匹配
Article.objects.filter(title='测试')  # 等价于 title__exact

# 大小写不敏感
Article.objects.filter(title__iexact='test')

# 包含
Article.objects.filter(title__contains='Django')
Article.objects.filter(title__icontains='django')  # 大小写不敏感

# 开头/结尾
Article.objects.filter(title__startswith='Django')
Article.objects.filter(title__endswith='教程')
Article.objects.filter(title__istartswith='django')

# 范围查询
Article.objects.filter(views__gt=100)   # 大于
Article.objects.filter(views__gte=100)  # 大于等于
Article.objects.filter(views__lt=100)   # 小于
Article.objects.filter(views__lte=100)  # 小于等于
Article.objects.filter(views__range=(10, 100))  # 范围

# IN 查询
Article.objects.filter(status__in=['draft', 'published'])
Article.objects.filter(pk__in=[1, 2, 3])

# NULL 查询
Article.objects.filter(category__isnull=True)

# 日期查询
Article.objects.filter(created_at__year=2024)
Article.objects.filter(created_at__month=6)
Article.objects.filter(created_at__day=15)
Article.objects.filter(created_at__week_day=1)  # 周一
Article.objects.filter(created_at__date=date.today())

# 正则查询
Article.objects.filter(title__regex=r'^Django.*教程$')
Article.objects.filter(title__iregex=r'^django')

3.4 跨关系查询

# 正向关系查询
Article.objects.filter(category__name='技术')
Article.objects.filter(author__email__endswith='@example.com')

# 反向关系查询
Category.objects.filter(articles__title__contains='Django')
User.objects.filter(articles__status='published').distinct()

# 多层关系
Article.objects.filter(author__profile__bio__contains='开发者')

3.5 Q 对象（复杂查询）

from django.db.models import Q

# OR 查询
articles = Article.objects.filter(
    Q(status='published') | Q(author=user)
)

# AND 查询
articles = Article.objects.filter(
    Q(status='published') & Q(views__gt=100)
)

# NOT 查询
articles = Article.objects.filter(
    ~Q(status='draft')
)

# 复杂组合
articles = Article.objects.filter(
    (Q(status='published') | Q(author=user)) &
    ~Q(category__isnull=True) &
    Q(created_at__year=2024)
)

# 动态构建查询
conditions = Q()
if status:
    conditions &= Q(status=status)
if author_id:
    conditions &= Q(author_id=author_id)
if keyword:
    conditions &= Q(title__icontains=keyword) | Q(content__icontains=keyword)

articles = Article.objects.filter(conditions)

3.6 F 对象（字段引用）

from django.db.models import F

# 字段比较
Article.objects.filter(updated_at__gt=F('created_at'))

# 字段运算
Article.objects.update(views=F('views') + 1)

# 关联字段
Article.objects.filter(author__username=F('category__name'))

# 数学运算
from django.db.models import ExpressionWrapper, DecimalField
Product.objects.annotate(
    discounted_price=ExpressionWrapper(
        F('price') * 0.8,
        output_field=DecimalField(max_digits=10, decimal_places=2)
    )
)

3.7 聚合与注解

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

# 聚合（返回字典）
result = Article.objects.aggregate(
    total=Count('id'),
    avg_views=Avg('views'),
    max_views=Max('views'),
    min_views=Min('views'),
)
# {'total': 100, 'avg_views': 150.5, 'max_views': 1000, 'min_views': 0}

# 注解（给每个对象添加计算字段）
categories = Category.objects.annotate(
    article_count=Count('articles'),
    total_views=Sum('articles__views'),
).filter(article_count__gt=5)

for category in categories:
    print(f"{category.name}: {category.article_count} 篇文章")

# 条件聚合
from django.db.models import Case, When, Value, IntegerField

Article.objects.aggregate(
    published_count=Count(
        Case(When(status='published', then=1))
    ),
    draft_count=Count(
        Case(When(status='draft', then=1))
    ),
)

# 分组统计
from django.db.models.functions import TruncMonth

Article.objects.annotate(
    month=TruncMonth('created_at')
).values('month').annotate(
    count=Count('id')
).order_by('month')

3.8 values 和 values_list

# 返回字典列表
articles = Article.objects.values('title', 'author__username')
# [{'title': '文章1', 'author__username': 'user1'}, ...]

# 返回元组列表
articles = Article.objects.values_list('title', 'views')
# [('文章1', 100), ('文章2', 200), ...]

# 返回单字段列表
titles = Article.objects.values_list('title', flat=True)
# ['文章1', '文章2', ...]

# 返回命名元组
from collections import namedtuple
articles = Article.objects.values_list('title', 'views', named=True)
# [Row(title='文章1', views=100), ...]

四、高级查询

4.1 select_related（一对一/一对多优化）

# 不优化：N+1 查询问题
articles = Article.objects.all()
for article in articles:
    print(article.author.username)  # 每次访问都查询数据库

# 使用 select_related：一次 JOIN 查询
articles = Article.objects.select_related('author', 'category')
for article in articles:
    print(article.author.username)  # 不再查询数据库

# 多层关联
articles = Article.objects.select_related(
    'author__profile',
    'category__parent',
)

4.2 prefetch_related（多对多优化）

# 不优化
articles = Article.objects.all()
for article in articles:
    for tag in article.tags.all():  # 每篇文章一次查询
        print(tag.name)

# 使用 prefetch_related：两次查询
articles = Article.objects.prefetch_related('tags')
for article in articles:
    for tag in article.tags.all():  # 使用缓存的数据
        print(tag.name)

# 自定义 Prefetch
from django.db.models import Prefetch

articles = Article.objects.prefetch_related(
    Prefetch(
        'tags',
        queryset=Tag.objects.filter(is_active=True).order_by('name'),
        to_attr='active_tags'  # 存储到自定义属性
    )
)

for article in articles:
    for tag in article.active_tags:  # 使用 to_attr
        print(tag.name)

4.3 Subquery 和 OuterRef

from django.db.models import Subquery, OuterRef

# 子查询
newest_article = Article.objects.filter(
    author=OuterRef('pk')
).order_by('-created_at')

users = User.objects.annotate(
    newest_article_title=Subquery(
        newest_article.values('title')[:1]
    )
)

# EXISTS 子查询
from django.db.models import Exists

published_articles = Article.objects.filter(
    author=OuterRef('pk'),
    status='published'
)

users = User.objects.annotate(
    has_published=Exists(published_articles)
).filter(has_published=True)

4.4 原生 SQL

# raw() 方法
articles = Article.objects.raw('''
    SELECT * FROM blog_article
    WHERE views > %s
    ORDER BY created_at DESC
''', [100])

# 执行自定义 SQL
from django.db import connection

with connection.cursor() as cursor:
    cursor.execute('''
        UPDATE blog_article
        SET views = views + 1
        WHERE id = %s
    ''', [article_id])

    cursor.execute('SELECT * FROM blog_article WHERE id = %s', [article_id])
    row = cursor.fetchone()

# 使用 extra（不推荐，已废弃）
articles = Article.objects.extra(
    select={'is_recent': 'created_at > NOW() - INTERVAL 7 DAY'},
    where=['views > %s'],
    params=[100],
)

4.5 数据库函数

from django.db.models.functions import (
    Lower, Upper, Length, Concat, Substr,
    Coalesce, Greatest, Least, Now,
    ExtractYear, ExtractMonth, TruncDate
)

# 字符串函数
Article.objects.annotate(
    title_lower=Lower('title'),
    title_length=Length('title'),
    full_title=Concat('category__name', Value(' - '), 'title'),
)

# 日期函数
Article.objects.annotate(
    year=ExtractYear('created_at'),
    month=ExtractMonth('created_at'),
    date=TruncDate('created_at'),
)

# Coalesce（返回第一个非空值）
Article.objects.annotate(
    display_name=Coalesce('nickname', 'username', Value('匿名'))
)

# Cast（类型转换）
from django.db.models.functions import Cast
from django.db.models import CharField

Article.objects.annotate(
    views_str=Cast('views', CharField(max_length=20))
)

五、模型继承

5.1 抽象基类

class TimestampMixin(models.Model):
    """时间戳混入类"""
    created_at = models.DateTimeField('创建时间', auto_now_add=True)
    updated_at = models.DateTimeField('更新时间', auto_now=True)

    class Meta:
        abstract = True  # 不创建数据库表

class Article(TimestampMixin):
    title = models.CharField('标题', max_length=200)
    content = models.TextField('内容')
    # 自动包含 created_at 和 updated_at

5.2 多表继承

class Place(models.Model):
    name = models.CharField('名称', max_length=100)
    address = models.CharField('地址', max_length=200)

class Restaurant(Place):
    serves_pizza = models.BooleanField('供应披萨', default=False)
    serves_pasta = models.BooleanField('供应意面', default=False)

# 创建两张表：place 和 restaurant
# restaurant 表包含指向 place 的外键

5.3 代理模型

class Article(models.Model):
    title = models.CharField('标题', max_length=200)
    status = models.CharField('状态', max_length=20)

class PublishedArticle(Article):
    """已发布文章的代理模型"""
    class Meta:
        proxy = True

    objects = PublishedManager()  # 自定义管理器

    def publish_to_social(self):
        """发布到社交媒体"""
        pass

# 使用相同的数据库表，但可以有不同的方法和管理器

六、模型管理器

6.1 自定义管理器

class ArticleQuerySet(models.QuerySet):
    def published(self):
        return self.filter(status='published')

    def by_author(self, user):
        return self.filter(author=user)

    def recent(self, days=7):
        from django.utils import timezone
        cutoff = timezone.now() - timezone.timedelta(days=days)
        return self.filter(created_at__gte=cutoff)

class ArticleManager(models.Manager):
    def get_queryset(self):
        return ArticleQuerySet(self.model, using=self._db)

    def published(self):
        return self.get_queryset().published()

    def by_author(self, user):
        return self.get_queryset().by_author(user)

class Article(models.Model):
    title = models.CharField('标题', max_length=200)
    status = models.CharField('状态', max_length=20)

    objects = ArticleManager()

    # 或使用 from_queryset
    # objects = ArticleManager.from_queryset(ArticleQuerySet)()

# 使用
Article.objects.published()
Article.objects.published().recent(days=30)
Article.objects.by_author(user).published()

6.2 多管理器

class PublishedManager(models.Manager):
    def get_queryset(self):
        return super().get_queryset().filter(status='published')

class Article(models.Model):
    title = models.CharField('标题', max_length=200)
    status = models.CharField('状态', max_length=20)

    objects = models.Manager()  # 默认管理器
    published = PublishedManager()  # 只返回已发布文章

# 使用
Article.objects.all()       # 所有文章
Article.published.all()     # 只有已发布文章

七、数据库迁移

7.1 基本迁移命令

# 创建迁移文件
python manage.py makemigrations
python manage.py makemigrations app_name

# 查看迁移 SQL
python manage.py sqlmigrate app_name 0001

# 执行迁移
python manage.py migrate
python manage.py migrate app_name

# 查看迁移状态
python manage.py showmigrations

# 回滚迁移
python manage.py migrate app_name 0001  # 回滚到指定版本
python manage.py migrate app_name zero  # 回滚所有

# 伪迁移
python manage.py migrate --fake app_name 0001

7.2 数据迁移

# 生成空迁移文件
# python manage.py makemigrations app_name --empty

from django.db import migrations

def forward_func(apps, schema_editor):
    Article = apps.get_model('blog', 'Article')
    for article in Article.objects.all():
        article.slug = slugify(article.title)
        article.save()

def reverse_func(apps, schema_editor):
    pass  # 回滚操作

class Migration(migrations.Migration):
    dependencies = [
        ('blog', '0001_initial'),
    ]

    operations = [
        migrations.RunPython(forward_func, reverse_func),
    ]

八、事务管理

8.1 atomic 装饰器

from django.db import transaction

@transaction.atomic
def create_article_with_tags(data):
    article = Article.objects.create(**data['article'])
    for tag_data in data['tags']:
        tag, _ = Tag.objects.get_or_create(**tag_data)
        article.tags.add(tag)
    return article

# 使用上下文管理器
def create_article_with_tags(data):
    with transaction.atomic():
        article = Article.objects.create(**data['article'])
        for tag_data in data['tags']:
            tag, _ = Tag.objects.get_or_create(**tag_data)
            article.tags.add(tag)
    return article

8.2 保存点

from django.db import transaction

def complex_operation():
    with transaction.atomic():
        article = Article.objects.create(title='测试')

        # 创建保存点
        sid = transaction.savepoint()

        try:
            # 可能失败的操作
            do_something_risky()
        except Exception:
            # 回滚到保存点
            transaction.savepoint_rollback(sid)
        else:
            # 提交保存点
            transaction.savepoint_commit(sid)

8.3 select_for_update

from django.db import transaction

@transaction.atomic
def transfer_funds(from_account_id, to_account_id, amount):
    # 锁定相关行，防止并发修改
    accounts = Account.objects.select_for_update().filter(
        id__in=[from_account_id, to_account_id]
    )

    from_account = accounts.get(id=from_account_id)
    to_account = accounts.get(id=to_account_id)

    if from_account.balance < amount:
        raise ValueError('余额不足')

    from_account.balance -= amount
    to_account.balance += amount

    from_account.save()
    to_account.save()

九、性能优化

9.1 批量操作

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

# 批量更新
Article.objects.filter(status='draft').update(status='published')

# bulk_update
articles = list(Article.objects.filter(status='draft'))
for article in articles:
    article.status = 'published'
    article.published_at = timezone.now()

Article.objects.bulk_update(articles, ['status', 'published_at'], batch_size=100)

9.2 迭代器

# 大数据集迭代
for article in Article.objects.iterator(chunk_size=1000):
    process(article)

# 或使用分页
from django.core.paginator import Paginator

articles = Article.objects.all()
paginator = Paginator(articles, 100)

for page_num in paginator.page_range:
    page = paginator.page(page_num)
    for article in page:
        process(article)

9.3 延迟加载

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

# defer：延迟加载指定字段
articles = Article.objects.defer('content')

# 组合使用
articles = Article.objects.select_related('author').only(
    'title', 'author__username'
)

9.4 索引优化

class Article(models.Model):
    title = models.CharField(max_length=200, db_index=True)
    status = models.CharField(max_length=20)
    created_at = models.DateTimeField()

    class Meta:
        indexes = [
            # 单列索引
            models.Index(fields=['status']),

            # 复合索引
            models.Index(fields=['status', '-created_at']),

            # 部分索引
            models.Index(
                fields=['title'],
                condition=Q(status='published'),
                name='published_title_idx'
            ),

            # 包含索引（PostgreSQL）
            models.Index(
                fields=['status'],
                include=['title', 'created_at'],
                name='status_covering_idx'
            ),
        ]

十、最佳实践

10.1 模型设计原则

# 1. 使用有意义的模型名和字段名
class Article(models.Model):  # 好
    pass

class Tbl001(models.Model):   # 差
    pass

# 2. 添加 verbose_name
class Article(models.Model):
    title = models.CharField('标题', max_length=200)  # 好

# 3. 定义 __str__ 方法
class Article(models.Model):
    def __str__(self):
        return self.title

# 4. 使用 related_name
author = models.ForeignKey(User, related_name='articles', on_delete=models.CASCADE)

# 5. 选择合适的 on_delete
category = models.ForeignKey(Category, on_delete=models.SET_NULL, null=True)

# 6. 避免在模型中使用复杂逻辑
# 使用服务层处理业务逻辑

10.2 查询优化清单

# 1. 使用 select_related 和 prefetch_related
articles = Article.objects.select_related('author').prefetch_related('tags')

# 2. 只查询需要的字段
articles = Article.objects.only('title', 'created_at')
articles = Article.objects.values('title', 'author__username')

# 3. 使用 exists() 而不是 count()
if Article.objects.filter(status='draft').exists():  # 好
    pass

if Article.objects.filter(status='draft').count() > 0:  # 差
    pass

# 4. 使用 update() 而不是循环保存
Article.objects.filter(status='draft').update(status='published')  # 好

# 5. 使用 bulk_create 和 bulk_update
Article.objects.bulk_create(articles)

# 6. 避免在循环中查询
# 差
for article in articles:
    category = Category.objects.get(id=article.category_id)

# 好
articles = Article.objects.select_related('category')
for article in articles:
    category = article.category

参考资源

• Django 模型官方文档
• QuerySet API 参考
• 数据库函数
• 迁移操作