Django 與數(shù)據(jù)庫交互，你需要知道的 9 個技巧

對開發(fā)人員來說，Django的ORM 確實非常實用，但是將數(shù)據(jù)庫的訪問抽象出來本身是有成本的，那些愿意在數(shù)據(jù)庫中探索的開發(fā)人員，經常會發(fā)現(xiàn)修改 ORM 的默認行為可以帶來性能的提升。在本文中，我將分享在 Django 中使用數(shù)據(jù)庫的 9 個技巧。

1. 過濾器聚合（Aggregation with Filter）

在 Django 2.0 之前，如果我們想要得到諸如用戶總數(shù)和活躍用戶總數(shù)之類的東西，我們不得不求助于條件表達式：

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 中，添加了聚合函數(shù)的過濾器參數(shù)，使其更容易：

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. 查詢集的結果變?yōu)榫呙M（QuerySet results as namedtuples）

我是一個 namedtuples 的粉絲，同時也是 Django 2.0 的 ORM 的粉絲。

在 Django 2.0 中，values_list?方法的參數(shù)中添加了一個叫做?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. 自定義函數(shù)（Custom functions）

Django 2.0 的 ORM 功能非常強大，而且特性豐富，但還是不能與所有數(shù)據(jù)庫的特性同步。不過幸運的是，ORM讓我們用自定義函數(shù)來擴展它。

假設我們有一個記錄報告的持續(xù)時間字段，我們希望找到所有報告的平均持續(xù)時間：

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 不支持間隔類型字段的求標準偏差操作，我們需要將時間間隔轉換為數(shù)字，然后才能對它應用?STDDEV_POP?操作。

一個選擇是從時間間隔中提?。?/span>

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

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

那么我們如何在 Django 中實現(xiàn)呢？你猜到了 -- 一個自定義函數(shù)：

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

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

我們的新函數(shù)這樣使用：

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 通常使用同步工作進程。這意味著，當用戶執(zhí)行長時間運行的操作時，工作進程會被阻塞，完成之前，其他人無法使用它。

應該沒有人真正在生產中只用一個工作進程來運行 Django，但是我們仍然希望確保一個查詢不會浪費太多資源太久。

在大多數(shù) Django 應用程序中，大部分時間都花在等待數(shù)據(jù)庫查詢上了。所以，在 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 任務等。

希望您使用的是持久的數(shù)據(jù)庫連接，這樣每次請求都不會再有連接開銷。超時也可以配置到用戶粒度：

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 行數(shù)據(jù)：

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

這很糟糕，因為雖然只返回 100 行數(shù)據(jù)，但是其實你已經把所有的行都取出來放進了內存。

我們再試試：

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

這個好多了，Django 會在 SQL 中使用 limit 子句來獲取 100 行數(shù)據(jù)。

我們增加了限制，但我們仍然有一個問題 -- 用戶想要所有的數(shù)據(jù)，但我們只給了他們 100 個，用戶現(xiàn)在認為只有 100 個數(shù)據(jù)了。

并非盲目的返回前 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. 事務與鎖的控制

這個比較難。由于數(shù)據(jù)庫中的鎖機制，我們開始在半夜發(fā)現(xiàn)事務超時錯誤。在我們的代碼中操作事務的常見模式如下所示：

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 并保存一些查詢。

更新交易還會涉及獲得一個鎖來確保它不被別人獲得。

現(xiàn)在，你看到問題了嗎？沒有？我也沒有。（作者好萌）

我們有一些晚上運行的 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）

創(chuàng)建模型時，Django 會在所有外鍵上創(chuàng)建一個 B-Tree 索引，它的開銷可能相當大，而且有時候并不很必要。

典型的例子是 M2M（多對多）關系的直通模型：

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

在上面的模型中，Django 將會隱式的創(chuàng)建兩個索引：一個用于用戶，一個用于組。

M2M 模型中的另一個常見模式是在兩個字段一起作為一個唯一約束。在這種情況下，意味著一個用戶只能是同一個組的成員，還是那個模型：

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

這個?unique_together?也會創(chuàng)建兩個索引，所以我們得到了兩個字段三個索引的模型 ?

根據(jù)我們用這個模型的職能，我們可以設置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',
        )

刪除冗余的索引將會是插入和查詢更快，而且我們的數(shù)據(jù)庫更輕量。

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

具有多個列的索引稱為組合索引。在 B-Tree 組合索引中，第一列使用樹結構進行索引。從第一層的樹葉為第二層創(chuàng)建一棵新樹，以此類推。

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

在上面的例子中，我們首先會得到一個組（group）的樹，另一個樹是所有它的用戶（user）。B-Tree 組合索引的經驗法則是使二級索引盡可能小。換句話說，高基數(shù)（更明確的值）的列應該是在第一位的。

在我們的例子中，假設組少于用戶（一般），所以把用戶列放在第一位會使組的二級索引變小。

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'順序調整了，使索引更小。

這只是一個經驗法則，最終的索引應該針對特定的場景進行優(yōu)化。這里的要點是要知道隱式索引和組合索引中列順序的重要性。

9. 塊范圍索引（BRIN indexes）

B-Tree 索引的結構像一棵樹。查找單個值的成本是隨機訪問表的樹的高度 + 1。這使得 B-Tree 索引非常適合獨特的約束和（一些）范圍查詢。

B-Tree索引的缺點是它的大小 -- B-Tree 索引可能會變大。

沒有其他選擇了嗎？并不是，數(shù)據(jù)庫為特定用例提供其他類型的索引也蠻多的。

從 Django 1.11 開始，有一個新的 Meta 選項用于在模型上創(chuàng)建索引。這給了我們探索其他類型索引的機會。

PostgreSQL 有一個非常有用的索引類型 BRIN（塊范圍索引）。在某些情況下，BRIN 索引可以比 B-Tree 索引更高效。

我們看看官網文檔怎么說的：

BRIN 設計用于處理非常大的表格，其中某些列與表格內的物理位置有一些自然的相關性。

要理解這個陳述，了解 BRIN 索引如何工作是很重要的。顧名思義，BRIN 索引會在表格中的一系列相鄰塊上創(chuàng)建一個小型索引。該索引非常小，只能說明某個值是否在范圍內，或者是否在索引塊范圍內。

我們來做一個 BRIN 索引如何幫助我們的簡單例子。

假設我們在一列中有這些值，每一個都是一個塊：

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

我們?yōu)槊咳齻€相鄰的塊創(chuàng)建一個范圍：

[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 索引的關鍵。為了充分利用它，列中的值必須大致排序或聚集在磁盤上。

現(xiàn)在回到 Django，我們有哪些常被索引的字段，最有可能在磁盤上自然排序？沒錯，就是?auto_now_add。（這個很常用，沒用到的小伙伴可以了解下）

Django 模型中一個非常常見的模式是：

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

當使用?auto_now_add?時，Django 將自動使用當前時間填充該行的時間。創(chuàng)建的字段通常也是查詢的絕佳候選字段，所以它通常被插入索引。

讓我們在創(chuàng)建時添加一個 BRIN 索引：

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

為了了解大小的差異，我創(chuàng)建了一個約 2M 行的表，并在磁盤上自然排序了日期字段：

• B-Tree 索引：37 MB

• BRIN 索引：49 KB

沒錯，你沒看錯。

創(chuàng)建索引時要考慮的要比索引的大小要多得多。但是現(xiàn)在，通過 Django 1.11 支持索引，我們可以輕松地將新類型的索引整合到我們的應用程序中，使它們更輕，更快。

原文地址：9 Django Tips for Working with Databases
原文作者：Haki Benita
譯者：臨書