python

Django 与数据库交互,你需要知道的 9 个点 - 知乎

文章暂存

systemime
2021-03-25
16 min

摘要.

此专栏我会不定期分享一些 Django 最前沿的文章,内容偏重技巧、经验的归纳总结,来源暂时有:

  1. Medium
  2. Twitter 知名博主

如果大家感兴趣,请一定点个关注,给我一些动力,毕竟翻译整理是需要时间的,谢谢大家

--

原文地址:9 Django Tips for Working with Databases
原文作者:Haki Benita
译者:临书
校对:临书
推荐程度:✨✨✨✨✨

对开发人员来说,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)

这与设置边界的最后一点有些相关。有时我们的客户的一些行为是不可预知的

比如,同一用户打开另一个选项卡并在第一次尝试「卡住」时再试一次并不罕见。

这就是为什么要限制

我们限制某一个查询的返回不超过 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_updateselect_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 也会创建两个索引,所以我们得到了个字段个索引的模型 ?

根据我们用这个模型的职能,我们可以忽略 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',
        )

*注意元组里面的字段名顺序

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


# 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]

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

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

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

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

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

再试着查找 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 支持索引,我们可以轻松地将新类型的索引整合到我们的应用程序中,使它们更轻,更快。

--

全文完,随意转载,转载请注明出处: Django 与数据库交互,你需要知道的 9 个点

觉得文章怎么样?我个人觉得是非常赞的!

如果大家觉得此类文章对你很有用,请务必点个关注,谢谢啦 https://zhuanlan.zhihu.com/p/33475891 https://zhuanlan.zhihu.com/p/33475891

上次编辑于: 2021/5/20 下午3:26:49