Django 与数据库交互，你需要知道的 9 个技巧-技术圈

对开发人员来说，Django的ORM 确实非常实用，但是将数据库的访问抽象出来本身是有成本的，那些愿意在数据库中探索的开发人员，经常会发现修改 ORM 的默认行为可以带来性能的提升。在本文中，我将分享在 Django 中使用数据库的 9 个技巧。

1. 过滤器聚合（Aggregation with Filter）

在 Django 2.0 之前，如果我们想要得到诸如用户总数和活跃用户总数之类的东西，我们不得不求助于条件表达式：

from django.contrib.auth.models import User
from django.db.models import (
    Count,
    Sum,
    Case,
    When,
    Value,
    IntegerField,
)

User.objects.aggregate(
    total_users=Count('id'),
    total_active_users=Sum(Case(
        When(is_active=True, then=Value(1)),
        default=Value(0),
        output_field=IntegerField(),
    )),
)

在 Django 2.0 中，添加了聚合函数的过滤器参数，使其更容易：

from django.contrib.auth.models import User
from django.db.models import Count, F

User.objects.aggregate(
    total_users=Count('id'),
    total_active_users=Count('id', filter=F('is_active')),
)

很棒，又短又可口。如果你正在使用 PostgreSQL，这两个查询将如下所示：

SELECT
    COUNT(id) AS total_users,
    SUM(CASE WHEN is_active THEN 1 ELSE 0 END) AS total_active_users
FROM
    auth_users;
SELECT
    COUNT(id) AS total_users,
    COUNT(id) FILTER (WHERE is_active) AS total_active_users
FROM
    auth_users;

第二个查询使用了 WHERE 过滤子句。

2. 查询集的结果变为具名元组（QuerySet results as namedtuples）

我是一个 namedtuples 的粉丝，同时也是 Django 2.0 的 ORM 的粉丝。

在 Django 2.0 中，values_list 方法的参数中添加了一个叫做 named 的属性。将 named 设置为 True 会将 QuerySet 作为 namedtuples 列表返回：

> user.objects.values_list(
    'first_name',
    'last_name',
)[0]
(‘Haki’, ‘Benita’)
> user_names = User.objects.values_list(
    'first_name',
    'last_name',
    named=True,
)
> user_names[0]
Row(first_name='Haki', last_name='Benita')
> user_names[0].first_name
'Haki'
> user_names[0].last_name
'Benita'

3. 自定义函数（Custom functions）

Django 2.0 的 ORM 功能非常强大，而且特性丰富，但还是不能与所有数据库的特性同步。不过幸运的是，ORM让我们用自定义函数来扩展它。

假设我们有一个记录报告的持续时间字段，我们希望找到所有报告的平均持续时间：

from django.db.models import Avg
Report.objects.aggregate(avg_duration=Avg(‘duration’))
> {'avg_duration': datetime.timedelta(0, 0, 55432)}

那很棒，但是如果只有均值，信息量有点少。我们再算出标准偏差吧：

from django.db.models import Avg, StdDev
Report.objects.aggregate(
    avg_duration=Avg('duration'),
    std_duration=StdDev('duration'),
)
ProgrammingError: function stddev_pop(interval) does not exist
LINE 1: SELECT STDDEV_POP("report"."duration") AS "std_dura...
               ^
HINT:  No function matches the given name and argument types.
You might need to add explicit type casts.

呃... PostgreSQL 不支持间隔类型字段的求标准偏差操作，我们需要将时间间隔转换为数字，然后才能对它应用 STDDEV_POP 操作。

一个选择是从时间间隔中提取：

SELECT
    AVG(duration),
    STDDEV_POP(EXTRACT(EPOCH FROM duration))
FROM 
    report;

      avg       |    stddev_pop
----------------+------------------
 00:00:00.55432 | 1.06310113695549
(1 row)

那么我们如何在 Django 中实现呢？你猜到了 -- 一个自定义函数：

# common/db.py
from django.db.models import Func

class Epoch(Func):
   function = 'EXTRACT'
   template = "%(function)s('epoch' from %(expressions)s)"

我们的新函数这样使用：

from django.db.models import Avg, StdDev, F
from common.db import Epoch

Report.objects.aggregate(
    avg_duration=Avg('duration'),
    std_duration=StdDev(Epoch(F('duration'))),
)
{'avg_duration': datetime.timedelta(0, 0, 55432),
 'std_duration': 1.06310113695549}

*注意在 Epoch 调用中使用 F 表达式。

4. 声明超时（Statement Timeout）

这可能是我给的最简单的也是最重要的提示。我们是人类，我们都会犯错。我们不可能考虑到每一个边缘情况，所以我们必须设定边界。

与其他非阻塞应用程序服务器（如 Tornado，asyncio 甚至 Node）不同，Django 通常使用同步工作进程。这意味着，当用户执行长时间运行的操作时，工作进程会被阻塞，完成之前，其他人无法使用它。

应该没有人真正在生产中只用一个工作进程来运行 Django，但是我们仍然希望确保一个查询不会浪费太多资源太久。

在大多数 Django 应用程序中，大部分时间都花在等待数据库查询上了。所以，在 SQL 查询上设置超时是一个很好的开始。

我喜欢像这样在我的 wsgi.py 文件中设置一个全局超时：

# wsgi.py
from django.db.backends.signals import connection_created
from django.dispatch import receiver

@receiver(connection_created)
def setup_postgres(connection, **kwargs):
    if connection.vendor != 'postgresql':
        return
    
    # Timeout statements after 30 seconds.
    with connection.cursor() as cursor:
        cursor.execute("""
            SET statement_timeout TO 30000;
        """)

为什么是 wsgi.py？因为这样它只会影响工作进程，不会影响进程外的分析查询，cron 任务等。

希望您使用的是持久的数据库连接，这样每次请求都不会再有连接开销。超时也可以配置到用户粒度：

postgresql=#> alter user app_user set statement_timeout TO 30000;
ALTER ROLE

题外话：我们花了很多时间在其他常见的地方，比如网络。因此，请确保在调用远程服务时始终设置超时时间：

import requests

response = requests.get(
    'https://api.slow-as-hell.com',
    timeout=3000,
)

5. 限制（Limit）

这与设置边界的最后一点有些相关。有时我们的客户的一些行为是不可预知的。比如，同一用户打开另一个选项卡并在第一次尝试「卡住」时再试一次并不罕见。

这就是为什么需要使用限制(Limit)。

我们限制某一个查询的返回不超过 100 行数据：

# bad example
data = list(Sale.objects.all())[:100]

这很糟糕，因为虽然只返回 100 行数据，但是其实你已经把所有的行都取出来放进了内存。

我们再试试：

data = Sale.objects.all()[:100]

这个好多了，Django 会在 SQL 中使用 limit 子句来获取 100 行数据。

我们增加了限制，但我们仍然有一个问题 -- 用户想要所有的数据，但我们只给了他们 100 个，用户现在认为只有 100 个数据了。

并非盲目的返回前 100 行，我们先确认一下，如果超过 100 行（通常是过滤以后），我们会抛出一个异常：

LIMIT = 100

if Sales.objects.count() > LIMIT:
    raise ExceededLimit(LIMIT)
return Sale.objects.all()[:LIMIT]

挺有用，但是我们增加了一个新的查询。

能不能做的更好呢？我们可以这样：

LIMIT = 100

data = Sale.objects.all()[:(LIMIT + 1)]
if len(data) > LIMIT:
    raise ExceededLimit(LIMIT)
return data

我们不取 100 行，我们取 100 + 1 = 101 行，如果 101 行存在，那么我们知道超过了 100 行：

记住 LIMIT + 1 窍门，有时候它会非常方便。

6. 事务与锁的控制

这个比较难。由于数据库中的锁机制，我们开始在半夜发现事务超时错误。在我们的代码中操作事务的常见模式如下所示：

from django.db import transaction as db_transaction

...
with db_transaction.atomic():
  transaction = (
        Transaction.objects
        .select_related(
            'user',
            'product',
            'product__category',
        )
        .select_for_update()
        .get(uid=uid)
  )
  ...

事务操作通常会涉及用户和产品的一些属性，所以我们经常使用 select_related 来强制 join 并保存一些查询。

更新交易还会涉及获得一个锁来确保它不被别人获得。

现在，你看到问题了吗？没有？我也没有。（作者好萌）

我们有一些晚上运行的 ETL 进程，主要是在产品和用户表上做维护。这些 ETL 操作会更新字段然后插入表，这样它们也会获得了表的锁。

那么问题是什么？当 select_for_update 与 select_related 一起使用时，Django 将尝试获取查询中所有表的锁。

我们用来获取事务的代码尝试获取事务表、用户、产品、类别表的锁。一旦 ETL 在午夜锁定了后三个表，交易就开始失败。

一旦我们对问题有了更好的理解，我们就开始寻找只锁定必要表（事务表）的方法。（又）幸运的是，select_for_update 的一个新选项在 Django 2.0 中可用：

from django.db import transaction as db_transaction

...
with db_transaction.atomic():
  transaction = (
        Transaction.objects
        .select_related(
            'user',
            'product',
            'product__category',
        )
        .select_for_update(
            of=('self',)
        )
        .get(uid=uid)
  )
  ...

这个 of 选项被添加到 select_for_update ，使用 of 可以指明我们要锁定的表，self 是一个特殊的关键字，表示我们要锁定我们正在处理的模型，即事务表。

目前，该功能仅适用于 PostgreSQL 和 Oracle。

7. 外键索引（FK Indexes）

创建模型时，Django 会在所有外键上创建一个 B-Tree 索引，它的开销可能相当大，而且有时候并不很必要。

典型的例子是 M2M（多对多）关系的直通模型：

class Membership(Model):
    group = ForeignKey(Group)
    user = ForeignKey(User)

在上面的模型中，Django 将会隐式的创建两个索引：一个用于用户，一个用于组。

M2M 模型中的另一个常见模式是在两个字段一起作为一个唯一约束。在这种情况下，意味着一个用户只能是同一个组的成员，还是那个模型：

class Membership(Model):
    group = ForeignKey(Group)
    user = ForeignKey(User)
    class Meta:
        unique_together = (
           'group',
           'user',
        )

这个 unique_together 也会创建两个索引，所以我们得到了两个字段三个索引的模型 ?

根据我们用这个模型的职能，我们可以设置db_index=False忽略 FK 索引，只保留唯一约束索引：

class Membership(Model):
    group = ForeignKey(Group, db_index=False)
    user = ForeignKey(User, db_index=False)
    class Meta:
        unique_together = (
            'group',
            'user',
        )

删除冗余的索引将会是插入和查询更快，而且我们的数据库更轻量。

8. 组合索引中列的顺序（Order of columns in composite index）

具有多个列的索引称为组合索引。在 B-Tree 组合索引中，第一列使用树结构进行索引。从第一层的树叶为第二层创建一棵新树，以此类推。

索引中列的顺序非常重要。

在上面的例子中，我们首先会得到一个组（group）的树，另一个树是所有它的用户（user）。B-Tree 组合索引的经验法则是使二级索引尽可能小。换句话说，高基数（更明确的值）的列应该是在第一位的。

在我们的例子中，假设组少于用户（一般），所以把用户列放在第一位会使组的二级索引变小。

class Membership(Model):
    group = ForeignKey(Group, db_index=False)
    user = ForeignKey(User, db_index=False)
    class Meta:
        unique_together = (
            'user',
            'group',
        )

*注意unique_together元组里面的'user'和'group'顺序调整了，使索引更小。

这只是一个经验法则，最终的索引应该针对特定的场景进行优化。这里的要点是要知道隐式索引和组合索引中列顺序的重要性。

9. 块范围索引（BRIN indexes）

B-Tree 索引的结构像一棵树。查找单个值的成本是随机访问表的树的高度 + 1。这使得 B-Tree 索引非常适合独特的约束和（一些）范围查询。

B-Tree索引的缺点是它的大小 -- B-Tree 索引可能会变大。

没有其他选择了吗？并不是，数据库为特定用例提供其他类型的索引也蛮多的。

从 Django 1.11 开始，有一个新的 Meta 选项用于在模型上创建索引。这给了我们探索其他类型索引的机会。

PostgreSQL 有一个非常有用的索引类型 BRIN（块范围索引）。在某些情况下，BRIN 索引可以比 B-Tree 索引更高效。

我们看看官网文档怎么说的：

BRIN 设计用于处理非常大的表格，其中某些列与表格内的物理位置有一些自然的相关性。

要理解这个陈述，了解 BRIN 索引如何工作是很重要的。顾名思义，BRIN 索引会在表格中的一系列相邻块上创建一个小型索引。该索引非常小，只能说明某个值是否在范围内，或者是否在索引块范围内。

我们来做一个 BRIN 索引如何帮助我们的简单例子。

假设我们在一列中有这些值，每一个都是一个块：

1, 2, 3, 4, 5, 6, 7, 8, 9

我们为每三个相邻的块创建一个范围：

[1,2,3], [4,5,6], [7,8,9]

对于每个范围，我们将保存范围内的最小值和最大值：

[1–3], [4–6], [7–9]

我们尝试通过此索引搜索 5：

[1–3] — 绝对没在这里
[4–6] — 可能在这里
[7–9] — 绝对没在这里

使用索引，我们限制了我们搜索的范围在 [4-6] 范围。

再举一个例子，这次列中的值不会被很好地排序：

[2–9], [1–7], [3–8]

再试着查找 5：

[2–9] — 可能在这里
[1–7] — 可能在这里
[3–8] — 可能在这里

索引是无用的 -- 它不仅没有限制搜索，实际上我们不得不搜索更多，因为我们同时提取了索引和整个表。

回到文档：

...列与表格内的物理位置有一些自然的相关性

这是 BRIN 索引的关键。为了充分利用它，列中的值必须大致排序或聚集在磁盘上。

现在回到 Django，我们有哪些常被索引的字段，最有可能在磁盘上自然排序？没错，就是 auto_now_add。（这个很常用，没用到的小伙伴可以了解下）

Django 模型中一个非常常见的模式是：

class SomeModel(Model):
    created = DatetimeField(
        auto_now_add=True,
    )

当使用 auto_now_add 时，Django 将自动使用当前时间填充该行的时间。创建的字段通常也是查询的绝佳候选字段，所以它通常被插入索引。

让我们在创建时添加一个 BRIN 索引：

from django.contrib.postgres.indexes import BrinIndex
class SomeModel(Model):
    created = DatetimeField(
        auto_now_add=True,
    )
    class Meta:
        indexes = (
            BrinIndex(fields=['created']),
        )

为了了解大小的差异，我创建了一个约 2M 行的表，并在磁盘上自然排序了日期字段：

B-Tree 索引：37 MB
BRIN 索引：49 KB

没错，你没看错。

创建索引时要考虑的要比索引的大小要多得多。但是现在，通过 Django 1.11 支持索引，我们可以轻松地将新类型的索引整合到我们的应用程序中，使它们更轻，更快。

原文地址：9 Django Tips for Working with Databases
原文作者：Haki Benita
译者：临书