在 Django 开发中使用数据库连接时需要了解的 9 个技巧

对于开发人员来说，ORM 非常实用，但数据库访问是有成本的。愿意看数据的开发人员，经常会发现改变ORM的标准行为可以提高其性能。

在本文中，我将分享在 Django 中使用数据库的 9 个技巧

1。带过滤器的集合

在 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;

第二个查询使用 ❀♸ 过滤器条款。

2。 QuerySet 结果作为名称元组

我是名称元组的粉丝，也是 Django 2.0 ORM 的粉丝。

在 Django 2.0 中，名为 的属性被添加到 values_list 方法的参数中。将名为文作的粘贴到True，它将返回查询集作为角色名称列表：

> 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。自定义函数

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。启示时间结束

这可能是我能给出的最简单也是最重要的提示。我们是人类，我们都会犯错误。我们无法计算所有的边缘情况，因此我们必须设置一个限制。

与 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？ 因为这种方法只会影响worker进程，不会影响进程外查询、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。边界（Boundaries）

这和最后一点关于设置边界有些关系。有时，某些客户端的行为是不可预测的

例如，同一用户打开另一个选项卡并重试而第一次尝试“卡住”是不常见的。

这就是为什么

我们限制请求，使数据不超过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_lated来强制访问并保存一些查询。

更新事务还包括获取锁以确保其他人无法访问它。

现在，你看到问题了吗？不？我也没有。 （作者很可爱）

我们有一些晚上运行的ETL流程，主要用于产品和用户表的维护。这些 ETL 作业更新字段并将其插入表中，因此它们也获取表锁。

那么问题出在哪里呢？当 select_for_update 与 select_lated 一起使用时，Django 将尝试获取查询中所有表的锁。

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

在我们对问题有了更好的了解之后，我们开始寻找一种方法来只关闭所需的表（销售表）。幸运的是（再次），Django 2.0 中提供了 select_for_update 的新选项：

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 选项已添加到

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

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

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

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

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

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

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

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

再举个例子，这次列中的值将不会被排序：

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

再次尝试查找 5：

• [2–9] — 这里有效 ❀ –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 树索引：37 MB
• BRIN 指数：49 KB

是的，您读到了。

创建索引时需要考虑的不仅仅是索引的大小。但现在，借助 Django 1.11 引用支持，我们可以轻松地向应用程序添加新类型的引用，使它们更轻、更快。