Django-优化权限验证
Python
Django-优化权限验证
Python

Django-优化权限验证

U1S1，权限看来看去感觉没啥看头，Django内置的就包含了组-用户-权限，可以针对组（角色）和用户单独控制，我觉得已经很强了。可能在复杂的业务中会涉及具体，但是对于目前的喵喵CRM来说，貌似还牵扯不到。

不过和某大佬聊天，问到权限验证。也就是某个用户访问后，系统到底怎么判断有没有权限的问题时候，才想到用户认证和权限验证这一块似乎有优化空间。

问题所在

目前采用的是JWT认证方式，每次用户操作产生如下步骤：

（1）用户（Apifox测试程序）客户端发送请求，假设是个POST创建新客户信息，请求中包含了Authorization：Bearer；

（2）服务器接收请求；

（3）JWT认证，DRF 的 JWT 认证中间件或认证类会截获这个请求。拿出 Token，验证签名和过期时间。验证通过后，它解析出 Token 里的用户 ID。执行第一次数据库查询，将这个用户ID对象找出来，然后创建一个request对象，将找出来的用户对象赋值给request.user；

（4）权限检查，视图代码开始执行，这儿应该会有一个权限检查，当检查时，应该会进行第二次数据库查询，Django的权限系统会去查询用户权限相关的表，把这个用户拥有的全部权限都查出来。这些查出来的权限被存入一个集合（set），然后缓存到当前这个 request.user里。

（5）假设用户具备创建新用户权限，权限检查通过。

（6）执行业务逻辑: 视图函数执行，在数据库里增加了一条新数据。

（7）返回响应: 服务器向 Apifox 返回一个 201 Created 的响应。

（8）生命周期结束。当响应发送完毕后，与这次请求相关的所有东西，包括 request 对象、request.user 对象以及它上面的权限缓存，全部被销毁和回收。服务器不保留任何关于这次请求的记忆。

所以问题就在于，用户每次操作都要去走一遍这个流程，无论是新增、修改、删除、查询还是啥的…高并发情况下数据库压力就比较大了。

问题验证

OK，眼见为实（这儿注意，DRF框架自带DjangoModelPermissions是不包含view权限的审查的，需要改一下参数）。

主程序下setting.py中DEBUG设置为True。

加个中间件便于测试：

from django.db import connection, reset_queries

class QueryCountMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        reset_queries()  # 每次请求前清空SQL记录
        response = self.get_response(request)
        num_queries = len(connection.queries)
        print(f"[SQL 查询次数] {request.path}: {num_queries}")
        for q in connection.queries:
            print(q['sql'], q['time'])  # 打印SQL语句
        return response

# 记得在setting.py中增加一下。

运行，使用Apifox访问两次用户列表，可以看到如下输出：

[SQL 查询次数] /permission/users/: 5
SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED 0.000

SELECT `auth_user`.`id`, `auth_user`.`password`, `auth_user`.`last_login`, `auth_user`.`is_superuser`, `auth_user`.`username`, `auth_user`.`first_name`, `auth_user`.`last_name`, `auth_user`.`email`, `auth_user`.`is_staff`, `auth_user`.`is_active`, `auth_user`.`date_joined` FROM `auth_user` WHERE `auth_user`.`id` = 1 LIMIT 21 0.001    

SELECT `django_content_type`.`app_label` AS `content_type__app_label`, `auth_permission`.`codename` AS `codename` FROM `auth_permission` INNER JOIN `auth_user_user_perm
issions` ON (`auth_permission`.`id` = `auth_user_user_permissions`.`permission_id`) INNER JOIN `django_content_type` ON (`auth_permission`.`content_type_id` = `django_content_type`.`id`) WHERE `auth_user_user_permissions`.`user_id` = 1 0.001

SELECT `django_content_type`.`app_label` AS `content_type__app_label`, `auth_permission`.`codename` AS `codename` FROM `auth_permission` INNER JOIN `auth_group_permissi
ons` ON (`auth_permission`.`id` = `auth_group_permissions`.`permission_id`) INNER JOIN `django_content_type` ON (`auth_permission`.`content_type_id` = `django_content_t

ype`.`id`) WHERE `auth_group_permissions`.`group_id` IN (SELECT U0.`id` FROM `auth_group` U0 INNER JOIN `auth_user_groups` U1 ON (U0.`id` = U1.`group_id`) WHERE U1.`user_id` = 1) 0.001
SELECT `auth_user`.`id`, `auth_user`.`password`, `auth_user`.`last_login`, `auth_user`.`is_superuser`, `auth_user`.`username`, `auth_user`.`first_name`, `auth_user`.`last_name`, `auth_user`.`email`, `auth_user`.`is_staff`, `auth_user`.`is_active`, `auth_user`.`date_joined` FROM `auth_user` 0.000
[25/Sep/2025 15:15:56] "GET /permission/users/ HTTP/1.1" 200 3310

[SQL 查询次数] /permission/users/: 5
SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED 0.000

SELECT `auth_user`.`id`, `auth_user`.`password`, `auth_user`.`last_login`, `auth_user`.`is_superuser`, `auth_user`.`username`, `auth_user`.`first_name`, `auth_user`.`last_name`, `auth_user`.`email`, `auth_user`.`is_staff`, `auth_user`.`is_active`, `auth_user`.`date_joined` FROM `auth_user` WHERE `auth_user`.`id` = 1 LIMIT 21 0.000    

SELECT `django_content_type`.`app_label` AS `content_type__app_label`, `auth_permission`.`codename` AS `codename` FROM `auth_permission` INNER JOIN `auth_user_user_permissions` ON (`auth_permission`.`id` = `auth_user_user_permissions`.`permission_id`) INNER JOIN `django_content_type` ON (`auth_permission`.`content_type_id` = `django_content_type`.`id`) WHERE `auth_user_user_permissions`.`user_id` = 1 0.001

SELECT `django_content_type`.`app_label` AS `content_type__app_label`, `auth_permission`.`codename` AS `codename` FROM `auth_permission` INNER JOIN `auth_group_permissions` ON (`auth_permission`.`id` = `auth_group_permissions`.`permission_id`) INNER JOIN `django_content_type` ON (`auth_permission`.`content_type_id` = `django_content_type`.`id`) WHERE `auth_group_permissions`.`group_id` IN (SELECT U0.`id` FROM `auth_group` U0 INNER JOIN `auth_user_groups` U1 ON (U0.`id` = U1.`group_id`) WHERE U1.`user_id` = 1) 0.001

SELECT `auth_user`.`id`, `auth_user`.`password`, `auth_user`.`last_login`, `auth_user`.`is_superuser`, `auth_user`.`username`, `auth_user`.`first_name`, `auth_user`.`last_name`, `auth_user`.`email`, `auth_user`.`is_staff`, `auth_user`.`is_active`, `auth_user`.`date_joined` FROM `auth_user` 0.000
[25/Sep/2025 15:15:59] "GET /permission/users/ HTTP/1.1" 200 3310

可以看到，每次访问都会完整走一遍查询的流程。Django自带的缓存只是存在单次请求里。

解决方案

python经典解决方案，遇到问题先装库，pip install django-redis。然后装一下redis。查了下，官方不支持，不过可以用docker。windows下docker安装不在赘述。安装好docker后，打开docker desktop，拉一个redis，docker run -d --name myredis -p 6379:6379 redis:latest映射下端口，然后可以用啦。

# setting.py 
# Redis配置
CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": "redis://127.0.0.1:6379/1",
        "OPTIONS": {
            "CLIENT_CLASS": "django_redis.client.DefaultClient",
        }
    }
}

上述配置好后，我们可以修改一下DRF原有的权限管理模块：


def get_user_perms(user):
    key = f"user:{user.id}:perms"
    perms = cache.get(key)
    if perms is None:
        perms = list(user.get_all_permissions())  # 如果缓存没有，就去数据库里查
        cache.set(key, perms, timeout=3600)  # 1小时过期，可调
    return perms


class CachedModelPermissions(DjangoModelPermissions):
    """
    等价于 DjangoModelPermissions，只是用缓存来减少数据库查询
    此外增加了 view相关权限的审查
    """
    perms_map = {
        "GET": ["%(app_label)s.view_%(model_name)s"],
        "OPTIONS": [],
        "HEAD": [],
        "POST": ["%(app_label)s.add_%(model_name)s"],
        "PUT": ["%(app_label)s.change_%(model_name)s"],
        "PATCH": ["%(app_label)s.change_%(model_name)s"],
        "DELETE": ["%(app_label)s.delete_%(model_name)s"],
    }

    def get_required_permissions(self, method, model_cls):
        """
        按照请求方法，返回所需的权限列表
        """
        return [perm % {
            "app_label": model_cls._meta.app_label,
            "model_name": model_cls._meta.model_name,
        } for perm in self.perms_map.get(method, [])]

    def has_permission(self, request, view):
        if not request.user or not request.user.is_authenticated:
            return False

        queryset = self._queryset(view)
        perms = get_user_perms(request.user)
        required_perms = self.get_required_permissions(request.method, queryset.model)

        # 只要缺一个权限就拒绝
        return all(perm in perms for perm in required_perms)

后续所有用到DjangoModelPermissions的全部改成CachedModelPermissions即可。

我们再来测试下：

[SQL 查询次数] /permission/users/: 5

SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED 0.000

SELECT `auth_user`.`id`, `auth_user`.`password`, `auth_user`.`last_login`, `auth_user`.`is_superuser`, `auth_user`.`username`, `auth_user`.`first_name`, `auth_user`.`last_name`, `auth_user`.`email`, `auth_user`.`is_staff`, `auth_user`.`is_active`, `auth_user`.`date_joined` FROM `auth_user` WHERE `auth_user`.`id` = 1 LIMIT 21 0.000    

SELECT `django_content_type`.`app_label` AS `content_type__app_label`, `auth_permission`.`codename` AS `codename` FROM `auth_permission` INNER JOIN `auth_user_user_perm
issions` ON (`auth_permission`.`id` = `auth_user_user_permissions`.`permission_id`) INNER JOIN `django_content_type` ON (`auth_permission`.`content_type_id` = `django_content_type`.`id`) WHERE `auth_user_user_permissions`.`user_id` = 1 0.001

SELECT `django_content_type`.`app_label` AS `content_type__app_label`, `auth_permission`.`codename` AS `codename` FROM `auth_permission` INNER JOIN `auth_group_permissi
ons` ON (`auth_permission`.`id` = `auth_group_permissions`.`permission_id`) INNER JOIN `django_content_type` ON (`auth_permission`.`content_type_id` = `django_content_t
ype`.`id`) WHERE `auth_group_permissions`.`group_id` IN (SELECT U0.`id` FROM `auth_group` U0 INNER JOIN `auth_user_groups` U1 ON (U0.`id` = U1.`group_id`) WHERE U1.`user_id` = 1) 0.001

SELECT `auth_user`.`id`, `auth_user`.`password`, `auth_user`.`last_login`, `auth_user`.`is_superuser`, `auth_user`.`username`, `auth_user`.`first_name`, `auth_user`.`last_name`, `auth_user`.`email`, `auth_user`.`is_staff`, `auth_user`.`is_active`, `auth_user`.`date_joined` FROM `auth_user` 0.000
[25/Sep/2025 15:58:42] "GET /permission/users/ HTTP/1.1" 200 3310

[SQL 查询次数] /permission/users/: 3

SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED 0.000

SELECT `auth_user`.`id`, `auth_user`.`password`, `auth_user`.`last_login`, `auth_user`.`is_superuser`, `auth_user`.`username`, `auth_user`.`first_name`, `auth_user`.`last_name`, `auth_user`.`email`, `auth_user`.`is_staff`, `auth_user`.`is_active`, `auth_user`.`date_joined` FROM `auth_user` WHERE `auth_user`.`id` = 1 LIMIT 21 0.000    

SELECT `auth_user`.`id`, `auth_user`.`password`, `auth_user`.`last_login`, `auth_user`.`is_superuser`, `auth_user`.`username`, `auth_user`.`first_name`, `auth_user`.`last_name`, `auth_user`.`email`, `auth_user`.`is_staff`, `auth_user`.`is_active`, `auth_user`.`date_joined` FROM `auth_user` 0.000
[25/Sep/2025 15:58:43] "GET /permission/users/ HTTP/1.1" 200 3310

可以看到，第二次同样的查询就只有用户认证+数据查询了，权限验证就不用再访问数据库去查了。不过权限如果修改，缓存如何更新可能还需要调整。而且缓存应用肯定可以放到更多的频繁查询的地方，后续再考虑考虑吧。

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注