django - 如何从 consumer.py django 访问用户信息？

Question

我正在使用 react.js 和 django、django-rest-framework、djoser、django-channels==3.0.2 构建一个社交媒体应用程序。问题是我想根据朋友请求等创建一个通知系统发布和评论。问题是我想要某种安全性，当我通过 websocket 从 react 发送请求时，我想检查发送者是否是经过身份验证的用户，通过在 websockets 上发送用户 ID，并将其与登录的在用户的 id 中。self.scope['user']由于我使用Djoser进行身份验证，所以我无法获得，还有其他方法可以实现吗？

Answer 1

您是否尝试在 django-channels 应用程序中添加中间件来验证和授权用户，如果没有，请按照以下步骤操作：1）在与 routing.py 相同的文件夹中创建一个名为 middleware.py 的新文件 2）添加以下内容middleware.py
注意：以下实现来自 django-channels 文档

from urllib.parse import parse_qs

from django.contrib.auth import get_user_model
from django.contrib.auth.models import AnonymousUser
from django.db import close_old_connections
from channels.auth import AuthMiddleware, AuthMiddlewareStack, UserLazyObject
from channels.db import database_sync_to_async
from channels.sessions import CookieMiddleware, SessionMiddleware
from rest_framework_simplejwt.tokens import AccessToken

User = get_user_model()

"""[summary]
plucks the JWT access token from the query string and retrieves the associated user.
  Once the WebSocket connection is opened, all messages can be sent and received without
  verifying the user again. Closing the connection and opening it again 
  requires re-authorization.
for example: 
ws://localhost:8000/<route>/?token=<token_of_the_user>

"""


@database_sync_to_async
def get_user(scope):
    close_old_connections()
    query_string = parse_qs(scope['query_string'].decode())
    token = query_string.get('token')
    if not token:
        return AnonymousUser()
    try:
        access_token = AccessToken(token[0])
        user = User.objects.get(id=access_token['id'])
    except Exception as exception:
        return AnonymousUser()
    if not user.is_active:
        return AnonymousUser()
    return user


class TokenAuthMiddleware(AuthMiddleware):
    async def resolve_scope(self, scope):
        scope['user']._wrapped = await get_user(scope)


def TokenAuthMiddlewareStack(inner):
    return CookieMiddleware(SessionMiddleware(TokenAuthMiddleware(inner)))

3）将中间件配置到路由中 3.1）打开routing.py

from channels.routing import ProtocolTypeRouter
from django.urls import path
from channels.routing import ProtocolTypeRouter, URLRouter
from <base-app-name>.middleware import TokenAuthMiddlewareStack

from <app-name>.consumers import <consumer-name>

application = ProtocolTypeRouter({
    'websocket': TokenAuthMiddlewareStack(
        URLRouter([
            path('<route>/', <consumer-name>),
        ]),
    ),
})

4) 在 AsyncWebSocketConsumer 的连接方法中检查用户

async def connect(self):
    user = self.scope['user']
    if user.is_anonymous:
        await self.close()
    else:
        <your logic>