DjangoのN+1問題はpytestで検知しよう

こんにちは！
デバイスソフトウエア開発部の米森です。担当している案件で、Webアプリ開発にDjangoを使用しています。

DjangoはPythonの「batteries included」という思想を踏襲しており、「バッテリー同梱なので、セットアップ作業無しですぐに色々な便利機能が使える」という考えをモットーの1つとして掲げています。

Webアプリ開発に必要な機能がほとんど全て揃っているようなDjangoですが、中でもORMが非常に強力で、SQLを意識することなく高速にWebアプリケーションを構築できます。しかし、その強力な抽象化の裏側で「いつの間にか意図せぬクエリが発行されていた」という恐怖を味わったことがあるエンジニアは多いのではないでしょうか。

その代表例が「N+1問題」です。

テストデータが少ない開発環境では問題が顕在化することが少ないので、動作確認の目をすり抜け、リリースされてしまうこともあります。

select_relatedや prefetch_relatedなどの回避策がDjangoから提供されていますが、これらが適切に使用されているかをコードレビューで確認するのには限界があります。

なので、今回はpytestを使ってN+1問題を機械的に検知する方法をご紹介します。

この記事で触れないこと

以下はこの記事では触れません。想定読者は、Django / DRF、pytestでの基本的な開発経験がある方です。

- Django/DRFを使ったWEBアプリケーションの作り方
- pytestの使い方
- Djangoとpytestの連携方法

前提：N+1問題とデータ構造

N+1問題とは、一覧取得のためのクエリ1回に加え、各レコードに紐づく関連データ取得のクエリがレコード数分（N回）発行されることで、結果として「N+1回」のクエリが実行されてしまう問題です。

以下の簡易的なデータ構造を想定します。

class User(models.Model):
    name = models.CharField(max_length=50)

    def __str__(self):
        return self.name

class Post(models.Model):
    author = models.ForeignKey(User, on_delete=models.CASCADE, related_name='posts')
    title = models.CharField(max_length=100)
    
    def __str__(self):
        return self.title

1

2

3

4

5

6

7

8

9

10

11

12

class User(models.Model):

name = models.CharField(max_length=50)

def __str__(self):

return self.name

class Post(models.Model):

author = models.ForeignKey(User, on_delete=models.CASCADE, related_name='posts')

title = models.CharField(max_length=100)

def __str__(self):

return self.title

PostモデルはUserモデルに紐づいており、Postの一覧を取得する際に、各Postのauthor.nameを表示するようなケースを例に考えます。

レスポンスイメージ：

[
    {
        "id": 1,
        "title": "Post 1",
        "author_name": "Alice"
    },
    {
        "id": 2,
        "title": "Post 2",
        "author_name": "Bob"
    },
    {
        "id": 3,
        "title": "Post 3",
        "author_name": "Charlie"
    }
]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

[

{

"id": 1,

"title": "Post 1",

"author_name": "Alice"

},

{

"id": 2,

"title": "Post 2",

"author_name": "Bob"

},

{

"id": 3,

"title": "Post 3",

"author_name": "Charlie"

}

]

例えば、Post 100件を取得する際、Post一覧の取得に1回、User情報の取得に100回、合計で101回のクエリが実行されのがN+1問題です。

データ量の増加に伴いクエリ数も増加するため、大規模データではパフォーマンス低下の原因となります。

まずは問題の再現

まずは、N+1問題が発生するケースを実際に再現してみます。

上述のmodelを使って、Postの一覧を取得するAPIを作成します。

PostモデルのModelSerializerを作成し、author.nameを取得するために、

serializers.ReadOnlyField(source='author.name')を使用します。

# serializers.py

from rest_framework import serializers
from .models import Post

class BadPostSerializer(serializers.ModelSerializer):
    author_name = serializers.ReadOnlyField(source='author.name') # !!! 関連テーブルアクセス

    class Meta:
        model = Post
        fields = ('id', 'title', 'author_name')

1

2

3

4

5

6

7

8

9

10

11

# serializers.py

from rest_framework import serializers

from .models import Post

class BadPostSerializer(serializers.ModelSerializer):

author_name = serializers.ReadOnlyField(source='author.name') # !!! 関連テーブルアクセス

class Meta:

model = Post

fields = ('id', 'title', 'author_name')

views.pyでは、Postの一覧を取得する際にPost.objects.all()を使用することで、関連テーブルであるUserへは事前にアクセスせず、各PostごとにUserへのクエリが発生するような書き方をします。

# views.py

from rest_framework import status
from rest_framework.response import Response
from rest_framework.views import APIView

from .models import Post
from .serializers import BadPostSerializer


class BadPostListAPIView(APIView):
    def get(self, request):
        posts = Post.objects.all() # !!! N+1問題が発生する可能性がある書き方
        serializer = BadPostSerializer(posts, many=True)
        data = serializer.data
        return Response(data, status=status.HTTP_200_OK)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

# views.py

from rest_framework import status

from rest_framework.response import Response

from rest_framework.views import APIView

from .models import Post

from .serializers import BadPostSerializer

class BadPostListAPIView(APIView):

def get(self, request):

posts = Post.objects.all() # !!! N+1問題が発生する可能性がある書き方

serializer = BadPostSerializer(posts, many=True)

data = serializer.data

return Response(data, status=status.HTTP_200_OK)

テストデータは以下のようなもので、Post一覧は10件のエントリを返します。

User(name="TestUser")
    ├ Post(title="01")
    ├ Post(title="02")
    ├ ....
    └ Post(title="10")

1

2

3

4

5

User(name="TestUser")

├ Post(title="01")

├ Post(title="02")

├ ....

└ Post(title="10")

その他、URLルーティングなどをセットアップし、APIを疎通させます。

実際に一覧取得を実行すると、以下のデータが取得できました。

想定通り、10件のPostが取得され、Postにはauthor.nameが含まれています。では、内部的にどのようなクエリが発行されているかを見てみます。 views.pyにクエリ情報をプリントします。

# views.py

class BadPostListAPIView(APIView):
    def get(self, request):
        posts = Post.objects.all()
        serializer = BadPostSerializer(posts, many=True)
        data = serializer.data
        ##### 以下追加
        queries = connection.queries
        query_count = len(queries)

        print("n================ [ N+1 Demo Query Log ] ================")
        print(f"Total Queries Executed: {query_count}")
        for i, q in enumerate(queries, 1):
            print(f"{i}. [{q['time']}s] {q['sql']}")
        print("========================================================n")        
        ##### 追加終わり
        return Response(data, status=status.HTTP_200_OK)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

# views.py

class BadPostListAPIView(APIView):

def get(self, request):

posts = Post.objects.all()

serializer = BadPostSerializer(posts, many=True)

data = serializer.data

##### 以下追加

queries = connection.queries

query_count = len(queries)

print("n================ [ N+1 Demo Query Log ] ================")

print(f"Total Queries Executed: {query_count}")

for i, q in enumerate(queries, 1):

print(f"{i}. [{q['time']}s] {q['sql']}")

print("========================================================n")

##### 追加終わり

return Response(data, status=status.HTTP_200_OK)

この状態でAPIを再度実行すると、以下のログが得られました。

Post一覧を取得するのに1回、それぞれのPostのauthor.nameを取得するのに10回、合計11回のクエリが発行されていることがわかります。

N+1問題をpytestで検知する

上記のような隠れたN+1問題を機械的に検知できるようにしたいです。

方法は色々あると思いますが、今回はpytest-djangoのdjango_assert_num_queriesを使う方法をご紹介します。

pytest-djangoはDjangoアプリケーション向けに便利なテスト機能を提供してくれるpytestのプラグインで、そのヘルパーの1つにdjango_assert_num_queries()というものがあります。

参考：https://pytest-django.readthedocs.io/en/latest/helpers.html#django_assert_num_queries

expected_query_num = 1
with django_assert_num_queries(expected_query_num):
    # テスト対象APIの実行
    response = client.get(url)

1

2

3

4

expected_query_num = 1

with django_assert_num_queries(expected_query_num):

# テスト対象APIの実行

response = client.get(url)

このように書くことで、テスト対象の処理で発行されるクエリ数をassertすることができます。

では実際に以下のようなテストコードを書いて、テストを実行してみます。

import pytest
from django.urls import reverse
from users.models import User, Post

@pytest.mark.django_db
def test_bad_post_list_query_count(client, django_assert_num_queries):
    ### Arrange
    # テスト用のダミーデータを作成 (User 1人、Post 3件)
    user1 = User.objects.create(name="Test User")

    Post.objects.create(author=user1, title="Post 1")
    Post.objects.create(author=user1, title="Post 2")
    Post.objects.create(author=user1, title="Post 3")

    url = reverse('bad-post-list')

    ### Act & Assert
    with django_assert_num_queries(1):
    # クエリが1回のみの発行であることを検証
        response = client.get(url)

    assert response.status_code == 200

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

import pytest

from django.urls import reverse

from users.models import User, Post

@pytest.mark.django_db

def test_bad_post_list_query_count(client, django_assert_num_queries):

### Arrange

# テスト用のダミーデータを作成 (User 1人、Post 3件)

user1 = User.objects.create(name="Test User")

Post.objects.create(author=user1, title="Post 1")

Post.objects.create(author=user1, title="Post 2")

Post.objects.create(author=user1, title="Post 3")

url = reverse('bad-post-list')

### Act & Assert

with django_assert_num_queries(1):

# クエリが1回のみの発行であることを検証

response = client.get(url)

assert response.status_code == 200

テスト結果はこちらです。

ちゃんとFAILしてくれましたね。クエリ発行回数の期待値が1に対して、4回発行されているので、テストが正常に動作しています。

FAILED tests/test_query_check.py::test_bad_post_list_query_count - Failed: Expected to perform 1 queries but 4 were done (add -v option to show queries)

1	FAILED tests/test_query_check.py::test_bad_post_list_query_count - Failed: Expected to perform 1 queries but 4 were done (add -v option to show queries)

コード側を改修

では、コード側に適切な実装を入れて、テストがPASSするようになるかを見てみます。今回のケースでは、views.pyでPost.objects.all()を使用している部分がN+1問題の原因になっています。

# views.py

class BadPostListAPIView(APIView):
    def get(self, request):
        posts = Post.objects.all() # !!! N+1問題が発生する可能性がある書き方

1

2

3

4

5

# views.py

class BadPostListAPIView(APIView):

def get(self, request):

posts = Post.objects.all() # !!! N+1問題が発生する可能性がある書き方

ここにselect_relatedを使って、PostとUserをJOINでまとめて取得するように書き換えます。

# views.py

class BadPostListAPIView(APIView):
    def get(self, request):
        # posts = Post.objects.all()
        posts = Post.objects.select_related('author').all() # NEW!
        serializer = BadPostSerializer(posts, many=True)
        data = serializer.data
        queries = connection.queries
        query_count = len(queries)

        print("n================ [ N+1 Demo Query Log ] ================")
        print(f"Total Queries Executed: {query_count}")
        for i, q in enumerate(queries, 1):
            print(f"{i}. [{q['time']}s] {q['sql']}")
        print("========================================================n")        
        return Response(data, status=status.HTTP_200_OK)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

# views.py

class BadPostListAPIView(APIView):

def get(self, request):

# posts = Post.objects.all()

posts = Post.objects.select_related('author').all() # NEW!

serializer = BadPostSerializer(posts, many=True)

data = serializer.data

queries = connection.queries

query_count = len(queries)

print("n================ [ N+1 Demo Query Log ] ================")

print(f"Total Queries Executed: {query_count}")

for i, q in enumerate(queries, 1):

print(f"{i}. [{q['time']}s] {q['sql']}")

print("========================================================n")

return Response(data, status=status.HTTP_200_OK)

まずはこの状態で、意図通りクエリが最適化されているかを確認します。 APIを実行すると、以下のレスポンスが返却されました。

改修前と同じレスポンスが得られているので、リファクタリングは成功していそうです。この時のクエリログはどうなっているのでしょうか。

必要なデータをINNER JOINを使うことで一括で取得できていそうです。では、最後にテストを再実行し、先ほどのテストケースがPASSするかを見ます。

無事PASSしました！これで安心してプッシュできそうです。

最後に

DjangoのORMは非常に便利ですが、N+1問題のようなパフォーマンス上の落とし穴がいたるところに潜んでいます。開発者の経験や勘に依存したコードレビューだけでは、こういった問題を見逃してしまうこともあります。今回ご紹介したツールなどを使って、機械的に検知する仕組みを作っておくと、リリース後のパフォーマンス問題を未然に防ぐことができますまた、一度土台を作れば、　生成AIを活用してテストケースを拡張することも可能です。

DjangoのN+1問題はpytestで検知しよう

この記事で触れないこと

前提：N+1問題とデータ構造

まずは問題の再現

N+1問題をpytestで検知する

コード側を改修

最後に

最近の投稿

カテゴリー

採用情報

IoTプラットフォーム「FASTIO」

この記事で触れないこと

前提：N+1問題とデータ構造

まずは問題の再現

N+1問題をpytestで検知する

コード側を改修

最後に

最近の投稿

人気の投稿

カテゴリー

採用情報

IoTプラットフォーム「FASTIO」