带有 querset User.objects.all() 的 ListAPIView 在测试数据库中仅对两个用户进行 7 次查询?
ListAPIView with the querset User.objects.all() makes 7 queries with only two users in the test database?
我正在使用 drf 创建端点来列出用户。在测试代码时,我意识到它调用了 7 个查询。
models.py:(我觉得用django的User模型会达到同样的效果)
class CustomUserManager(BaseUserManager):
"""
Custom user model manager where email is the unique identifiers
for authentication instead of usernames.
"""
def create_user(self, email, password, **extra_fields):
"""
Create and save a User with the given email and password.
"""
if not email:
raise ValueError(_("The Email must be set"))
email = self.normalize_email(email)
user = self.model(email=email, **extra_fields)
user.set_password(password)
user.save()
return user
def create_superuser(self, email, password, **extra_fields):
"""
Create and save a SuperUser with the given email and password.
"""
extra_fields.setdefault("is_staff", True)
extra_fields.setdefault("is_superuser", True)
extra_fields.setdefault("is_active", True)
if extra_fields.get("is_staff") is not True:
raise ValueError(_("Superuser must have is_staff=True."))
if extra_fields.get("is_superuser") is not True:
raise ValueError(_("Superuser must have is_superuser=True."))
return self.create_user(email, password, **extra_fields)
class BaseUser(AbstractUser):
email = models.EmailField(_("email address"), unique=True)
id_number = models.CharField(max_length=MID_LENGTH)
username = None
USERNAME_FIELD = "email"
REQUIRED_FIELDS = [email]
objects = CustomUserManager()
def __str__(self) -> str:
return self.email
serializers.py:
class UserSerializer(serializers.ModelSerializer):
class Meta:
model = models.BaseUser
fields = "__all__"
api.py:
class ListUserView(ListAPIView):
permission_classes = [IsAdminUser]
queryset = models.BaseUser.objects.all().order_by("id")
serializer_class = serializers.UserSerializer
test.py:
from . import models
from django.db import connection
from django.test.utils import CaptureQueriesContext
from django.urls import reverse
from django_seed import Seed
from rest_framework.test import APIClient, APITestCase
seeder = Seed.seeder()
CREDENTIALS = ["test@email.com", "12345678"]
class UserViewsTest(APITestCase):
def setUp(self) -> None:
# Create user and login
self.user = models.BaseUser.objects.create(email=CREDENTIALS[0], is_staff=True)
self.user.set_password(CREDENTIALS[1])
self.user.save()
self.assertTrue(
self.client.login(email=CREDENTIALS[0], password=CREDENTIALS[1])
)
# Seed users
seeder.add_entity(models.BaseUser, 1)
seeder.execute()
def tearDown(self) -> None:
self.client.logout()
def test_list_users(self):
"""
7 queries:
1. Logged in user session
2. 2 of each of:
i. 1 query for base user
ii. 1 query for base user's group
iii. 1 query for base user's permission
"""
with CaptureQueriesContext(connection) as queries:
response = self.client.get(reverse("accounts-list-users"))
self.assertEqual(200, response.status_code)
self.assertIsNotNone(response.content)
print(queries.captured_queries)
它进行 7 次查询的原因可能是什么?我知道一个查询是针对会话内容的。此外,有什么方法可以减少查询次数?
下面是 print(queries.captured_queries) 的输出:
[
{
"sql": "SELECT `django_session`.`session_key`, `django_session`.`session_data`, `django_session`.`expire_date` FROM `django_session` WHERE (`django_session`.`expire_date` > '2022-02-22 08:41:16.815976' AND `django_session`.`session_key` = 'hebjioxebumhz0y1mumtafwk3lfoj81h') LIMIT 21",
"time": "0.001",
},
{
"sql": "SELECT `accounts_baseuser`.`id`, `accounts_baseuser`.`password`, `accounts_baseuser`.`last_login`, `accounts_baseuser`.`is_superuser`, `accounts_baseuser`.`first_name`, `accounts_baseuser`.`last_name`, `accounts_baseuser`.`is_staff`, `accounts_baseuser`.`is_active`, `accounts_baseuser`.`date_joined`, `accounts_baseuser`.`email`, `accounts_baseuser`.`id_number` FROM `accounts_baseuser` WHERE `accounts_baseuser`.`id` = 1 LIMIT 21",
"time": "0.000",
},
{
"sql": "SELECT `accounts_baseuser`.`id`, `accounts_baseuser`.`password`, `accounts_baseuser`.`last_login`, `accounts_baseuser`.`is_superuser`, `accounts_baseuser`.`first_name`, `accounts_baseuser`.`last_name`, `accounts_baseuser`.`is_staff`, `accounts_baseuser`.`is_active`, `accounts_baseuser`.`date_joined`, `accounts_baseuser`.`email`, `accounts_baseuser`.`id_number` FROM `accounts_baseuser` ORDER BY `accounts_baseuser`.`id` ASC",
"time": "0.000",
},
{
"sql": "SELECT `auth_group`.`id`, `auth_group`.`name` FROM `auth_group` INNER JOIN `accounts_baseuser_groups` ON (`auth_group`.`id` = `accounts_baseuser_groups`.`group_id`) WHERE `accounts_baseuser_groups`.`baseuser_id` = 1",
"time": "0.001",
},
{
"sql": "SELECT `auth_permission`.`id`, `auth_permission`.`name`, `auth_permission`.`content_type_id`, `auth_permission`.`codename` FROM `auth_permission` INNER JOIN `accounts_baseuser_user_permissions` ON (`auth_permission`.`id` = `accounts_baseuser_user_permissions`.`permission_id`) INNER JOIN `django_content_type` ON (`auth_permission`.`content_type_id` = `django_content_type`.`id`) WHERE `accounts_baseuser_user_permissions`.`baseuser_id` = 1 ORDER BY `django_content_type`.`app_label` ASC, `django_content_type`.`model` ASC, `auth_permission`.`codename` ASC",
"time": "0.001",
},
{
"sql": "SELECT `auth_group`.`id`, `auth_group`.`name` FROM `auth_group` INNER JOIN `accounts_baseuser_groups` ON (`auth_group`.`id` = `accounts_baseuser_groups`.`group_id`) WHERE `accounts_baseuser_groups`.`baseuser_id` = 2",
"time": "0.000",
},
{
"sql": "SELECT `auth_permission`.`id`, `auth_permission`.`name`, `auth_permission`.`content_type_id`, `auth_permission`.`codename` FROM `auth_permission` INNER JOIN `accounts_baseuser_user_permissions` ON (`auth_permission`.`id` = `accounts_baseuser_user_permissions`.`permission_id`) INNER JOIN `django_content_type` ON (`auth_permission`.`content_type_id` = `django_content_type`.`id`) WHERE `accounts_baseuser_user_permissions`.`baseuser_id` = 2 ORDER BY `django_content_type`.`app_label` ASC, `django_content_type`.`model` ASC, `auth_permission`.`codename` ASC",
"time": "0.000",
},
]
7 个查询:
- 获取会话。为避免此查询,您可以使用例如the
cached_db
session backend.
- 从数据库中获取登录用户的详细信息。这也可以通过不同的会话后端从缓存(或者可能是会话对象本身)获取数据来避免(但是你必须小心,例如缓存失效)。
- 列出用户。由于您要列出用户,这是可以预料的。
- 获取用户 1 的分配组。
- 获取用户 1 的分配权限。
- 获取用户 2 的分配组。
- 获取用户 2 的分配权限。
可以通过以下方式避免查询 4 到 7:
- 正在从序列化程序中删除数据。 (您当前正在使用
fields = "__all__"
。)如果您的序列化程序未序列化权限和组,则也不会获取它们。
- 向视图集的查询集添加合适的
.prefetch_related()
(因为这些是 M2M)子句,因此这些是在 2 个查询中完成的,而不是 2N 个查询。
我正在使用 drf 创建端点来列出用户。在测试代码时,我意识到它调用了 7 个查询。
models.py:(我觉得用django的User模型会达到同样的效果)
class CustomUserManager(BaseUserManager):
"""
Custom user model manager where email is the unique identifiers
for authentication instead of usernames.
"""
def create_user(self, email, password, **extra_fields):
"""
Create and save a User with the given email and password.
"""
if not email:
raise ValueError(_("The Email must be set"))
email = self.normalize_email(email)
user = self.model(email=email, **extra_fields)
user.set_password(password)
user.save()
return user
def create_superuser(self, email, password, **extra_fields):
"""
Create and save a SuperUser with the given email and password.
"""
extra_fields.setdefault("is_staff", True)
extra_fields.setdefault("is_superuser", True)
extra_fields.setdefault("is_active", True)
if extra_fields.get("is_staff") is not True:
raise ValueError(_("Superuser must have is_staff=True."))
if extra_fields.get("is_superuser") is not True:
raise ValueError(_("Superuser must have is_superuser=True."))
return self.create_user(email, password, **extra_fields)
class BaseUser(AbstractUser):
email = models.EmailField(_("email address"), unique=True)
id_number = models.CharField(max_length=MID_LENGTH)
username = None
USERNAME_FIELD = "email"
REQUIRED_FIELDS = [email]
objects = CustomUserManager()
def __str__(self) -> str:
return self.email
serializers.py:
class UserSerializer(serializers.ModelSerializer):
class Meta:
model = models.BaseUser
fields = "__all__"
api.py:
class ListUserView(ListAPIView):
permission_classes = [IsAdminUser]
queryset = models.BaseUser.objects.all().order_by("id")
serializer_class = serializers.UserSerializer
test.py:
from . import models
from django.db import connection
from django.test.utils import CaptureQueriesContext
from django.urls import reverse
from django_seed import Seed
from rest_framework.test import APIClient, APITestCase
seeder = Seed.seeder()
CREDENTIALS = ["test@email.com", "12345678"]
class UserViewsTest(APITestCase):
def setUp(self) -> None:
# Create user and login
self.user = models.BaseUser.objects.create(email=CREDENTIALS[0], is_staff=True)
self.user.set_password(CREDENTIALS[1])
self.user.save()
self.assertTrue(
self.client.login(email=CREDENTIALS[0], password=CREDENTIALS[1])
)
# Seed users
seeder.add_entity(models.BaseUser, 1)
seeder.execute()
def tearDown(self) -> None:
self.client.logout()
def test_list_users(self):
"""
7 queries:
1. Logged in user session
2. 2 of each of:
i. 1 query for base user
ii. 1 query for base user's group
iii. 1 query for base user's permission
"""
with CaptureQueriesContext(connection) as queries:
response = self.client.get(reverse("accounts-list-users"))
self.assertEqual(200, response.status_code)
self.assertIsNotNone(response.content)
print(queries.captured_queries)
它进行 7 次查询的原因可能是什么?我知道一个查询是针对会话内容的。此外,有什么方法可以减少查询次数?
下面是 print(queries.captured_queries) 的输出:
[
{
"sql": "SELECT `django_session`.`session_key`, `django_session`.`session_data`, `django_session`.`expire_date` FROM `django_session` WHERE (`django_session`.`expire_date` > '2022-02-22 08:41:16.815976' AND `django_session`.`session_key` = 'hebjioxebumhz0y1mumtafwk3lfoj81h') LIMIT 21",
"time": "0.001",
},
{
"sql": "SELECT `accounts_baseuser`.`id`, `accounts_baseuser`.`password`, `accounts_baseuser`.`last_login`, `accounts_baseuser`.`is_superuser`, `accounts_baseuser`.`first_name`, `accounts_baseuser`.`last_name`, `accounts_baseuser`.`is_staff`, `accounts_baseuser`.`is_active`, `accounts_baseuser`.`date_joined`, `accounts_baseuser`.`email`, `accounts_baseuser`.`id_number` FROM `accounts_baseuser` WHERE `accounts_baseuser`.`id` = 1 LIMIT 21",
"time": "0.000",
},
{
"sql": "SELECT `accounts_baseuser`.`id`, `accounts_baseuser`.`password`, `accounts_baseuser`.`last_login`, `accounts_baseuser`.`is_superuser`, `accounts_baseuser`.`first_name`, `accounts_baseuser`.`last_name`, `accounts_baseuser`.`is_staff`, `accounts_baseuser`.`is_active`, `accounts_baseuser`.`date_joined`, `accounts_baseuser`.`email`, `accounts_baseuser`.`id_number` FROM `accounts_baseuser` ORDER BY `accounts_baseuser`.`id` ASC",
"time": "0.000",
},
{
"sql": "SELECT `auth_group`.`id`, `auth_group`.`name` FROM `auth_group` INNER JOIN `accounts_baseuser_groups` ON (`auth_group`.`id` = `accounts_baseuser_groups`.`group_id`) WHERE `accounts_baseuser_groups`.`baseuser_id` = 1",
"time": "0.001",
},
{
"sql": "SELECT `auth_permission`.`id`, `auth_permission`.`name`, `auth_permission`.`content_type_id`, `auth_permission`.`codename` FROM `auth_permission` INNER JOIN `accounts_baseuser_user_permissions` ON (`auth_permission`.`id` = `accounts_baseuser_user_permissions`.`permission_id`) INNER JOIN `django_content_type` ON (`auth_permission`.`content_type_id` = `django_content_type`.`id`) WHERE `accounts_baseuser_user_permissions`.`baseuser_id` = 1 ORDER BY `django_content_type`.`app_label` ASC, `django_content_type`.`model` ASC, `auth_permission`.`codename` ASC",
"time": "0.001",
},
{
"sql": "SELECT `auth_group`.`id`, `auth_group`.`name` FROM `auth_group` INNER JOIN `accounts_baseuser_groups` ON (`auth_group`.`id` = `accounts_baseuser_groups`.`group_id`) WHERE `accounts_baseuser_groups`.`baseuser_id` = 2",
"time": "0.000",
},
{
"sql": "SELECT `auth_permission`.`id`, `auth_permission`.`name`, `auth_permission`.`content_type_id`, `auth_permission`.`codename` FROM `auth_permission` INNER JOIN `accounts_baseuser_user_permissions` ON (`auth_permission`.`id` = `accounts_baseuser_user_permissions`.`permission_id`) INNER JOIN `django_content_type` ON (`auth_permission`.`content_type_id` = `django_content_type`.`id`) WHERE `accounts_baseuser_user_permissions`.`baseuser_id` = 2 ORDER BY `django_content_type`.`app_label` ASC, `django_content_type`.`model` ASC, `auth_permission`.`codename` ASC",
"time": "0.000",
},
]
7 个查询:
- 获取会话。为避免此查询,您可以使用例如the
cached_db
session backend. - 从数据库中获取登录用户的详细信息。这也可以通过不同的会话后端从缓存(或者可能是会话对象本身)获取数据来避免(但是你必须小心,例如缓存失效)。
- 列出用户。由于您要列出用户,这是可以预料的。
- 获取用户 1 的分配组。
- 获取用户 1 的分配权限。
- 获取用户 2 的分配组。
- 获取用户 2 的分配权限。
可以通过以下方式避免查询 4 到 7:
- 正在从序列化程序中删除数据。 (您当前正在使用
fields = "__all__"
。)如果您的序列化程序未序列化权限和组,则也不会获取它们。 - 向视图集的查询集添加合适的
.prefetch_related()
(因为这些是 M2M)子句,因此这些是在 2 个查询中完成的,而不是 2N 个查询。